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)[源代码]#

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.

示例

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)
参数:
run(message, *, config=None, model=None, reasoning_effort=None, parallel_tool_call=None, metadata=None, tags=None)[源代码]#

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

参数:
返回类型:

AssistantMessage

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

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.

参数:
返回类型:

SyncStream[AgentEvent]

resume_point(*, config=None, model=None)[源代码]#

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

参数:
返回类型:

ResumePoint | None

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

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

参数:
返回类型:

AssistantMessage

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

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

参数:
返回类型:

SyncStream[AgentEvent]

emit_external_event(event, *, config=None)[源代码]#

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

参数:
返回类型:

list[str]

close()[源代码]#

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

返回类型:

None