> ## Documentation Index
> Fetch the complete documentation index at: https://docs.cycls.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Custom loops

> Hook the default loop with a few lines, or replace it entirely and keep the session, provider, tools and compaction.

Most agents never need this page. The default loop handles streaming, tools,
retries, compaction and persistence. Reach for a custom loop when your product
needs orchestration the loop does not model, such as routing between models,
running a planning pass, or enforcing a state machine.

## Hooking without replacing

Events are dicts, so you can observe and react while still passing everything
through.

```python theme={null}
async for ev in llm.run(context=context):
    if isinstance(ev, dict):
        if ev.get("type") == "step" and ev.get("tool_name") == "web_search":
            metrics.increment("search")
        if ev.get("type") == "callout" and ev.get("style") == "error":
            alert_ops(ev["callout"])
    yield ev
```

You can also transform or drop events, or inject your own around them.

```python theme={null}
yield {"type": "status", "status": "Working"}
async for ev in llm.run(context=context):
    yield ev
yield {"type": "ui", "action": "suggest", "text": "Turn this into a report"}
```

## Replacing the loop

```python theme={null}
from cycls._agent.harness import (
    default_loop, make_provider, Session, build_tools, dispatch, compact, events,
)


async def my_loop(*, context, system="", tools=None, allowed_tools=(), model=None, **kw):
    provider = make_provider(model)
    session = await Session.open(context)
    await session.add_user(context.messages.raw[-1].get("content", ""))

    tool_list = build_tools(allowed_tools, tools or [], vendor=model.split("/", 1)[0])

    async for ev in provider.stream(
        messages=session.context(),
        system=system,
        tools=tool_list,
        max_tokens=kw.get("max_tokens") or 8192,
    ):
        yield ev

    await session.checkpoint()


llm = cycls.LLM().model("anthropic/claude-sonnet-4-6").loop(my_loop)
```

`fn` is an async generator that yields events. Accept `**kw` so new keyword
arguments do not break your loop as the SDK grows.

## The kit

| Piece                                                | What it does                                                           |
| ---------------------------------------------------- | ---------------------------------------------------------------------- |
| `default_loop`                                       | the built-in loop, callable directly if you want to wrap it            |
| `make_provider(model, ...)`                          | one streaming interface over Anthropic and OpenAI-compatible endpoints |
| `Session`                                            | the message log and its persistence                                    |
| `build_tools(allowed, custom, vendor=, web_search=)` | the provider-neutral tool list                                         |
| `dispatch(block, workspace, timeout, handlers, ...)` | run one tool call, returns a step event and a coroutine                |
| `compact(...)`                                       | token-budgeted context window with an append-only marker               |
| `events`                                             | dict factories for `text`, `thinking`, `step`, `callout`, `tool_call`  |
| `ToolContext`                                        | who a tool acts for: user, workspace, chat, approvals                  |

## Session

`Session` is the part worth understanding, because it is what makes a run
survivable.

```python theme={null}
session = await Session.open(context)

await session.add_user(content)     # checkpoints the user turn before any model call
session.messages                    # the working message list
session.context()                   # the window the provider should see
await session.checkpoint()          # persist what has happened so far
session.rollback()                  # drop the assistant tail after a failure
await session.compact(provider)     # summarize older turns
```

`add_user` writes to disk before the model is called, so a dropped connection
never loses what the person said. `rollback` removes only the assistant tail, so
a failed turn does not corrupt the transcript.

## The provider interface

```python theme={null}
class Provider(Protocol):
    model: str

    def stream(self, *, messages, system, tools, max_tokens,
               mcp_servers=None, thinking=None) -> AsyncIterator: ...

    async def complete(self, *, messages, system, max_tokens) -> str: ...
```

`stream` yields UI events, bare strings for text deltas, then exactly one `Turn`
carrying the assistant content, stop reason and token counts. Adding a new
provider means one file that conforms to this protocol.

The message shape is Anthropic's, which is the richest superset across vendors.
Each provider translates from it to its own wire format.

## What you give up

Writing your own loop means owning the behavior the default one provides:
retries with backoff, compaction before overflow, one recovery replay on a
context error, checkpointing between tool batches, heartbeats during long tool
runs, cost logging, and unwinding cleanly on cancellation.

Start by wrapping `default_loop` and intercepting what you need, then replace it
only when wrapping stops being enough.

## Next

<Card title="Interface" icon="window" href="/web/interface">
  Branding, themes, the explore menu and the example gallery.
</Card>