Custom loops

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.

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.

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

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.

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

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.

Interface

Branding, themes, the explore menu and the example gallery.

Get started

Functions

Apps

Agents

Interface and users

Ship it

Guides

Reference

Resources

For LLMs

Custom loops

Hooking without replacing

Replacing the loop

The kit

Session

The provider interface

What you give up

Next

Interface

​Hooking without replacing

​Replacing the loop

​The kit

​Session

​The provider interface

​What you give up

​Next

Interface

Hooking without replacing

Replacing the loop

The kit

Session

The provider interface

What you give up

Next