Use it from synchronous code#

Not every program is async. create_agent_sync runs the same runtime on one background event loop, so scripts, CLIs, and notebooks can call it from ordinary def code. Model behavior, tools, extensions, session history, retries, and event types are identical to the async API.

Run and stream#

from zett_agent.sync import create_agent_sync

with create_agent_sync(model) as agent:
    reply = agent.run("Hello")
    print(reply.content)

    with agent.stream("Continue") as events:
        for event in events:
            print(event.type, event.delta)

Use the stream as a context manager if you might stop early; closing it cancels the unfinished request and waits for cleanup. Later requests on the same session run normally.

Blocking views of async objects#

Agents, clients, providers, tools, storage, and dispatchers expose .sync(), which mirrors their async methods with blocking calls:

with existing_agent.sync() as agent:
    agent.initialize(config=config)
    reply = agent.run("Hello")

with read_file.sync() as read:
    result = read({"path": "README.md"})

with storage.sync() as db:
    session = db.load("session-42")

For anything else — free functions, third-party clients — use SyncRuntime directly:

with SyncRuntime() as runtime:
    agent = runtime.call(Agent.create, model, config=config)
    result = runtime.call(agent.run, "Hello")
    with runtime.stream(agent.stream, "Continue") as events:
        for event in events:
            print(event)

Share one runtime for resources that belong together, and close what you own before leaving it. A runtime you pass in explicitly stays open when a view exits.

Callbacks and custom models#

Extension hooks and dispatcher callbacks stay async def; they run on the background loop with the real request context. GUI applications should schedule visual updates through their own UI thread rather than touching widgets from a callback.

If your model is a blocking def stream(request) generator, wrap it with SyncModelAdapter. Existing @tool functions already support both styles.

Sending input into a running request#

While a synchronous stream is paused at an Ask User or Plan Mode event, another thread can deliver the reply with agent.emit_external_event(event, config=config). Different sessions can run concurrently; overlapping requests for one session are rejected.

Complete offline example#

"""Run, stream, and persist a conversation without an asynchronous entry point."""

from pathlib import Path
from tempfile import TemporaryDirectory

from zett_agent.agent import AgentRunConfig
from zett_agent.dispatcher import AgentEventDispatcher
from zett_agent.events import AgentEventType
from zett_agent.extensions.sqlite import SQLiteSessionExtension
from zett_agent.messages import (
    AssistantMessage,
    UserMessage,
)
from zett_agent.model import (
    ModelEvent,
    ModelResponse,
)
from zett_agent.sync import (
    SyncModelAdapter,
    create_agent_sync,
)


class Echo:
    """A synchronous model used only to make this example runnable offline."""

    def stream(self, request):
        users = [message for message in request.messages if isinstance(message, UserMessage)]
        answer = f"Turn {len(users)}: {users[-1].text}"
        yield ModelEvent.text(answer)
        yield ModelEvent.completed(ModelResponse(AssistantMessage(content=answer)))


class Console(AgentEventDispatcher):
    async def on_text_delta_event(self, event):
        print(event.delta)


def main() -> None:
    with TemporaryDirectory() as directory:
        history = SQLiteSessionExtension(Path(directory) / "sessions.sqlite3")
        try:
            with create_agent_sync(
                SyncModelAdapter(Echo()),
                config=AgentRunConfig("sync-example"),
                extensions=[history],
                event_dispatcher=Console(),
            ) as agent:
                assert agent.run("Hello").content == "Turn 1: Hello"
                with agent.stream("Remember this") as events:
                    for event in events:
                        if event.type is AgentEventType.RUN_COMPLETED:
                            assert event.message is not None
                            assert event.message.content == "Turn 2: Remember this"
            # Storage is asynchronous; a plain function reaches it through the sync view.
            with history.storage.sync() as database:
                records = database.list_raw_messages("sync-example")
                assert [record.message.role for record in records] == [
                    "system",
                    "user",
                    "assistant",
                    "system",
                    "user",
                    "assistant",
                ]
            print("Synchronous history persisted")
        finally:
            with history.storage.sync() as database:
                database.close()


if __name__ == "__main__":
    main()

Download the example.