Agent Protocol¶

latent.protocol 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 every consumer aligned on one set of names and one type identity per event.

from latent.protocol import (
    Message,                 # conversation history element
    AgentEvent,              # base class of the event hierarchy
    TextDelta, ReasoningDelta,
    LLMCallStart, LLMCallEnd,
    ToolCall, ToolResult, FunctionCall,
    RetrieverStart, RetrieverEnd,
    StepBoundary, Usage, Error,
    Step, InvokeResult,      # buffered view returned by invoke()
    AgentProtocol,           # the contract itself (runtime-checkable)
)

Where the definitions live

The underlying types are defined in latent.events (the event hierarchy + Message) and latent.agents.invoke (Step, InvokeResult). Both modules still work, but new code should import from latent.protocol.

The contract¶

Every agent implementation satisfies AgentProtocol. It has two required entry points and one optional one:

Method	Kind	Returns	Use it for
`stream(messages, *, config=None)`	async iterator	`AsyncIterator[AgentEvent]`	Per-event observation — live UIs, OTEL tracers, eval harnesses
`invoke(messages, *, stream_wrapper=None, config=None)`	coroutine	`InvokeResult`	Per-step structured consumption — batch runners, Temporal activities, dataset replays
`reset()` (optional, duck-typed)	sync	`None`	Clearing a transient in-memory convenience buffer (see below)

messages is the full conversation history; the last message is the new user input. config is an optional dict — consumers MAY pass otel_context for agent-internal tracing or conversation_id for correlation.

The contract is stateless

Agents carry no conversation state across calls — each stream() / invoke() is a pure function of the messages you pass in. That's what lets one agent instance serve many conversations concurrently, and why runtimes can replay datasets through it. Persisting history is the consumer's / platform's responsibility (e.g. Agent Studio). reset() is optional and only clears a transient in-memory buffer that convenience helpers like ReActAgent.ask() keep for REPL-style turn-by-turn use — it is not a state-management API.

AgentProtocol is @runtime_checkable, so you can verify an object's shape. Every agent shipped in the package satisfies it — including PipelineAgent, which inherits stream() / invoke() from BaseAgent:

from latent.protocol import AgentProtocol
from latent.agents import ReActAgent

agent = ReActAgent(name="qa", model="gpt-4o")
assert isinstance(agent, AgentProtocol)   # checks for stream()/invoke()

`Message` — conversation history¶

@dataclass
class Message:
    role: str
    content: str | list[dict[str, Any]]
    id: str | None = None
    tool_calls: list[ToolCall] | None = None
    tool_call_id: str | None = None

The field layout matches LiteLLM's ChatCompletionMessage verbatim — no internal-shape translation. Two roles are special:

An assistant message that invoked tools sets tool_calls to a list of ToolCall instances — the same dataclass the agent stream yields.
A tool message that carries one result sets role="tool", tool_call_id="<matching id>", and content to the stringified output. One role="tool" row per result.

The event hierarchy¶

stream() yields AgentEvent subclasses in emission order. The wire types:

Event	Fields	Emitted
`TextDelta`	`text`	Each chunk of customer-facing text
`ReasoningDelta`	`text`	Each chunk of reasoning/thinking content
`LLMCallStart`	`run_id`, `input_preview`, `model_id`, `timestamp`	Before each LLM API call
`LLMCallEnd`	`run_id`, `output_preview`, `input_tokens`, `output_tokens`, `model_id`, `timestamp`	After each LLM API call
`ToolCall`	`id`, `function` (`FunctionCall`), `type`	The LLM invoked a tool
`ToolResult`	`tool_call_id`, `output`, `error`	A tool finished executing
`RetrieverStart`	`run_id`, `query`, `name`, `timestamp`	A `@retriever` query began
`RetrieverEnd`	`run_id`, `documents`, `name`, `timestamp`	A `@retriever` query completed
`StepBoundary`	`step`	Start of each agent-loop iteration
`Usage`	`input_tokens`, `output_tokens`, `model_id`, `cache_creation_input_tokens`, `cache_read_input_tokens`	Token usage for one LLM call
`Error`	`message`	Terminal failure (e.g. max iterations exceeded)

Specialized agents emit additional markers — GuidelineMatched, JourneyStepAdvanced, JourneyCompleted (guided agents) and PhaseStarted, PhaseCompleted, PhaseRouted (pipeline agents).

from latent.agents import ReActAgent
from latent.protocol import Message, TextDelta, ToolCall

agent = ReActAgent(name="demo", model="gpt-4o", tools=[my_tool])

async for event in agent.stream([Message(role="user", content="Use the tool")]):
    match event:
        case TextDelta():
            print(event.text, end="", flush=True)
        case ToolCall():
            print(f"\n→ {event.function.name}({event.function.arguments})")

`StepBoundary` segments the stream¶

StepBoundary is emitted exactly once at the start of each loop iteration, before any other event for that iteration. Everything after it (until the next StepBoundary or end-of-stream) belongs to that step. This is the segmentation primitive invoke() uses to build its per-step view, so subclasses that override stream() must preserve the invariant.

The buffered view: `Step` and `InvokeResult`¶

invoke() consumes stream() internally, segments it on StepBoundary, and returns an InvokeResult:

@dataclass
class Step:
    step: int
    reasoning: str = ""
    text: str = ""
    tool_calls: list[ToolCall] = field(default_factory=list)
    tool_messages: list[Message] = field(default_factory=list)   # role="tool"

@dataclass
class InvokeResult:
    text: str            # terminal step's text — the customer-facing answer
    steps: list[Step]    # full per-iteration breakdown

result.text is derived structurally from the terminal step after the consumption loop completes — never from a live buffer cleared on a boundary signal. Non-terminal step text is preserved in steps (for debugging / dataset views) but excluded from text.

result = await agent.invoke([Message(role="user", content="Use the tool, then answer")])

print(result.text)                 # final answer
for step in result.steps:
    print(step.reasoning)          # thinking for this iteration
    print(step.tool_calls)         # list[ToolCall] the LLM invoked
    print(step.tool_messages)      # list[Message] role="tool" results

`stream_wrapper`¶

External observers (OTEL tracers, sinks) inject themselves into invoke() without owning the consumption loop by passing a stream_wrapper — a callable that wraps the internal event stream before consumption:

from latent.protocol import StreamWrapper  # Callable[[AsyncIterator], AsyncIterator]

Tool-call shape¶

All three surfaces that carry tool calls — Message.tool_calls (history), Step.tool_calls (buffered view), and the ToolCall event — use the same dataclass, whose layout matches LiteLLM's ChatCompletionAssistantToolCall:

@dataclass
class FunctionCall:
    name: str
    arguments: str          # JSON-encoded STRING (parse with json.loads)

@dataclass
class ToolCall(AgentEvent):
    id: str
    function: FunctionCall
    type: str = "function"

arguments is a JSON string, not a dict — the wire format every provider uses through LiteLLM. Build one from native values with ToolCall.from_arguments(id=..., name=..., arguments={"x": 1}).

A tool's result is a plain Message, not a combined call+result record — the invoke() / history path stores the assistant's tool calls and each tool result as separate rows:

Message(role="tool", tool_call_id="<id>", content=str(output))

So persisting a Step back into LLM history is a pure copy — no translation:

history.append(Message(role="assistant", tool_calls=step.tool_calls))
history.extend(step.tool_messages)

Observability¶

The event stream is the substrate for agent-level observability — metrics and sinks consume the same AgentEvents documented above, no separate hooks needed.

`collect_call_metrics` → `CallMetrics`¶

Aggregate token and latency metrics from a list of AgentEvents (for example, the events you collected from stream()). It sums every LLM call and derives duration from matched LLMCallStart/LLMCallEnd timestamp pairs:

from latent.agents import collect_call_metrics, CallMetrics
from latent.protocol import Message

events = [event async for event in agent.stream([Message(role="user", content="hi")])]

metrics: CallMetrics = collect_call_metrics(events)
print(metrics.duration_seconds)   # float — summed across all LLM calls
print(metrics.input_tokens)       # int
print(metrics.output_tokens)      # int
print(metrics.total_tokens)       # int — input + output

Token counts come from the LLMCallEnd.input_tokens / output_tokens fields (the same per-call accounting carried by the Usage events documented above).

Sinks: `AgentEventSink` + `AgentSinkMiddleware`¶

AgentEventSink is a runtime-checkable Protocol — anything with an async def emit(self, event: AgentEvent) -> None is a sink. AgentSinkMiddleware wraps any agent so every streamed event is mirrored to each sink while passing through unchanged. Sinks are observe-only; a failing sink is logged, never raised. These are agent-level observability sinks — distinct from the guardrail Scanners and their pass/block semantics.

from latent.agents.observability.sink_middleware import AgentSinkMiddleware
from latent.agents.observability.sinks import AgentEventSink, StructlogSink, OpenTelemetrySink
from latent.protocol import Message

wrapped = AgentSinkMiddleware(
    agent,
    sinks=[StructlogSink(), OpenTelemetrySink(max_text_chars=500)],
)

async for event in wrapped.stream([Message(role="user", content="hi")]):
    ...  # every TextDelta, ToolCall, LLMCallEnd, Usage … flows to both sinks

StructlogSink(logger_name="latent.agents", level=logging.INFO, max_text_chars=500) logs each event as a structured record.
OpenTelemetrySink(max_text_chars=500) emits each event as a span event on the active span — a no-op when opentelemetry-api is absent or no span is recording.

Roadmap: `serve()` (planned)¶

TBD — coming to AgentProtocol

A third entry point, serve(), is planned. Where stream() and invoke() run a single turn, agent.serve() will expose an agent as a long-lived service — accepting many conversations over a transport (e.g. HTTP / WebSocket) while keeping the same stateless contract (every request carries its own messages).

The exact signature is still being designed, so this section is a placeholder; it will be filled in when serve() lands. Until then, run an agent as a service by wrapping invoke() / stream() behind your own server endpoint.

Agent Protocol¶

The contract¶

Message — conversation history¶

The event hierarchy¶

StepBoundary segments the stream¶

The buffered view: Step and InvokeResult¶

stream_wrapper¶