SyncAgent#

Understand request ownership, phases, and runtime limits. See the guide.

class zett_agent.sync.SyncAgent(model, *, config=None, system_prompt='You are a helpful assistant.', tools=(), extensions=None, reasoning_effort=ReasoningEffort.MEDIUM, parallel_tool_call=True, max_iterations=36, max_internal_messages=8, event_dispatcher=None, runtime=None)[source]#

An initialized Agent callable from ordinary synchronous Python code.

All requests share one event loop so SDK connections and session resources survive across calls. Different sessions may be used by different calling threads. The underlying Agent still rejects overlapping requests within one session. Provider and extension resources remain caller-owned.

The synchronous interface does not change Extension execution: lifecycle hooks remain async and receive the original AgentRunContext on the background loop. A supplied runtime is shared and is not closed when this view exits.

Examples

Use a supplied model, persistence, and synchronous event handlers:

with SyncAgent(model, extensions=[history]) as agent:
    answer = agent.run("Hello")
    with agent.stream("Continue") as events:
        for event in events:
            print(event.type, event.delta)

Share a scope when a provider also needs explicit cleanup:

with SyncRuntime() as runtime:
    provider = runtime.call(DeepSeekProvider, model="deepseek-chat")
    try:
        with SyncAgent(provider, runtime=runtime) as agent:
            print(agent.run("Hello").content)
    finally:
        runtime.call(provider.aclose)
Parameters:
run(message, *, config=None, model=None, reasoning_effort=None, parallel_tool_call=None, metadata=None, tags=None)[source]#

Return a final answer, preserving original errors and event callbacks.

Parameters:
Return type:

AssistantMessage

stream(message, *, config=None, model=None, reasoning_effort=None, parallel_tool_call=None, metadata=None, tags=None)[source]#

Return a lazy event iterator; the request starts on the first next().

Use with agent.stream(…) as events when breaking early so cancellation and cleanup finish immediately. A plain for loop does not close an iterator on break. The same AgentEvent types as Agent.stream are yielded.

Parameters:
Return type:

SyncStream[AgentEvent]

resume_point(*, config=None, model=None)[source]#

Report where the session was interrupted, or None when there is nothing to resume.

Parameters:
Return type:

ResumePoint | None

resume(*, config=None, model=None, reasoning_effort=None, parallel_tool_call=None, metadata=None, tags=None)[source]#

Pick an interrupted session up where it stopped and return the final answer.

Parameters:
Return type:

AssistantMessage

stream_resume(*, config=None, model=None, reasoning_effort=None, parallel_tool_call=None, metadata=None, tags=None)[source]#

Return a lazy event iterator for a resume; it starts on the first next().

Parameters:
Return type:

SyncStream[AgentEvent]

emit_external_event(event, *, config=None)[source]#

Route Ask User, Plan Mode, and steering replies while generation waits.

Parameters:
Return type:

list[str]

close()[source]#

Close the owned runtime; an explicitly supplied runtime stays open.

Return type:

None