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.

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.

Next

Interface

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