SyncModelAdapter

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

class zett_agent.sync.SyncModelAdapter(model)

Use a model whose stream(request) returns an ordinary synchronous iterator.

Iteration and cleanup run in workers, so a blocking SDK never freezes the Agent event loop. The wrapped model and its transport remain caller-owned. Synchronous next() cannot be forcibly terminated; cancellation waits for the current call to return. Configure SDK transport timeouts accordingly.

示例

Implement an entirely synchronous teaching model:

class Echo:
    def stream(self, request):
        yield ModelEvent.text("Hello")
        yield ModelEvent.completed(
            ModelResponse(AssistantMessage(content="Hello"))
        )

with SyncAgent(SyncModelAdapter(Echo())) as agent:
    print(agent.run("Hi").content)

参数:

model (Any)

async stream(request)

Translate one blocking iterator into the existing async model protocol.

参数:

request (ModelRequest)

返回类型:

AsyncIterator[ModelEvent]

sync(*, runtime=None)

Return a context-managed synchronous view; optionally share a runtime.

Omitting runtime creates a new background thread for this view. Passing an existing runtime reuses its loop and leaves its shutdown to the caller. This API is for service entry points, not Extension lifecycle hooks.

示例

Use a model or tool from a plain function:

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

参数:

runtime (SyncRuntime | None)

返回类型:

SyncObject[Self]