Skip to content

latent.protocol

Canonical agent protocol for the Latent stack.

This module is the single import surface for everyone programming against the agent contract — agents that produce events, runtimes that consume them, frameworks that wrap them, and tools that observe them.

Importing from here keeps consumers aligned on one set of names and one type identity for each kind of event. The underlying definitions live in latent.events (event hierarchy + Message) and latent.agents.invoke (buffered per-step view). Both modules continue to work for backward compatibility, but new code should prefer latent.protocol.

What's here:

  • Wire types every agent emits: :class:Message, :class:AgentEvent and its 11 subclasses (:class:TextDelta, :class:ReasoningDelta, :class:LLMCallStart / :class:LLMCallEnd, :class:ToolCall / :class:ToolResult, :class:StepBoundary, :class:Usage, :class:Error, and the journey/phase markers).
  • Buffered types the :meth:AgentProtocol.invoke API returns: :class:Step, :class:InvokeResult.
  • Tool-call shape — :class:ToolCall and :class:FunctionCall dataclasses whose field layout matches LiteLLM's ChatCompletionAssistantToolCall verbatim. The same :class:ToolCall instance plays two roles: the streaming event the agent yields, and the value type stored on historical assistant messages. Tool results are plain :class:Message instances with role="tool" — see below.
  • The contract itself: :class:AgentProtocol — @runtime_checkable so consumers can isinstance(agent, AgentProtocol) to verify shape.
  • A convenience alias: :data:StreamWrapper for the stream_wrapper parameter of invoke().

Tool-call shape — locked down to OpenAI / LiteLLM. All three surfaces in the protocol that carry tool calls use the same dict shape that LiteLLM uses across every provider:

  • :attr:Message.tool_calls (history) — list[ToolCall] | None
  • :attr:Step.tool_calls (buffered view) — same
  • :attr:Step.tool_messages (results) — list[Message] with each entry's role="tool" and tool_call_id set, mirroring OpenAI's split between the assistant turn and its tool results. Same :class:Message type as conversation history — persisting a step into history is a pure list.extend.

No internal "combined call+result" record — historical ToolCallRecord was removed in 5.4.0. Persisting a step back into LLM history is a copy: Message(role="assistant", tool_calls=step.tool_calls) followed by history.extend(step.tool_messages).

Architectural note: the protocol intentionally sits in latent (the framework) rather than agent-studio (the platform that consumes it). The dependency direction matches reality — agents are framework implementations; platforms wrap them.

Classes

AgentProtocol

AgentProtocol()

Contract every agent implementation must satisfy.

Two required entry points and one optional one:

  • :meth:stream — yields :class:AgentEvent items in emission order. The wire-level API. Every consumer that wants per-event observation (live UIs, OTEL tracers, eval harnesses) calls this.
  • :meth:invoke — buffered consumption that returns an :class:InvokeResult. The structured API. Every consumer that wants the per-step view (Temporal activities, batch runners, dataset replays) calls this.
  • :meth:reset (optional, duck-typed) — clears agent state between conversations. Stateful agents (in-memory history, LangGraph thread state) implement it; stateless agents can omit it.

@runtime_checkable enables isinstance(my_agent, AgentProtocol) for shape verification — it checks for the presence of the methods (not their signatures). Contract tests in tests/unit/test_protocol_adherence.py use this to verify every shipped agent in this package satisfies the protocol.

Methods

AgentProtocol.invoke

invoke(messages: list[Message], stream_wrapper: StreamWrapper | None = None, config: dict[str, Any] | None = None) -> InvokeResult

Buffered call — return per-step structured result.

Internally consumes :meth:stream, builds the per-step :class:Step list, and derives the terminal-step text structurally from steps. See :class:InvokeResult for the guaranteed shape.

Args: messages: Full conversation history. stream_wrapper: Optional callable that wraps the internal stream before consumption. External observers (OTEL tracers, sinks) inject themselves here without owning the consumption loop. config: Optional configuration dictionary (same semantics as :meth:stream).

Returns: :class:InvokeResult with the terminal-step text in result.text and the full per-step structure in result.steps.

AgentProtocol.stream

stream(messages: list[Message], config: dict[str, Any] | None = None) -> AsyncIterator[AgentEvent]

Stream agent events for the given conversation history.

Args: messages: Full conversation history. The last message is the new user input. config: Optional configuration dictionary. Consumers MAY pass otel_context for agent-internal tracing or conversation_id for correlation.

Yields: :class:AgentEvent items as the agent processes the input.

Attributes

StreamWrapper