Each tablestake, the code that uses it, and the demo that shows it. Run any
demo with go run ./examples/agentkit-demo <name>.
llm.Client.ChatStream returns a channel of StreamChunk (content deltas,
then tool calls, then a final chunk carrying Usage). Content streams token by
token; tool-call fragments are reassembled internally before emission.
Rate limits are handled inside the client. On 429 it waits and retries,
honoring the server's retry_after — from the Retry-After header or a
JSON backpressure body like corrallm's:
{"error":{"reason":"queue-timeout","retry_after":10,"type":"backpressure"}}429 retries until a slot frees, bounded by Client.RetryBudget (default 5m) and
your ctx deadline, whichever is shorter — on a contended endpoint another wait
is another shot at a slot, but the budget stops it hanging forever. 5xx retries
a bounded number of times then fails. A retry delay is always clamped to a
safety ceiling.
client := llm.NewClient(baseURL, apiKey, model)
client.RetryBudget = 5 * time.Minute // cap total time spent retrying one requestch, err := client.ChatStream(ctx, msgs, tools, &llm.ChatOpts{TraceID: "thread-42"})
for chunk := range ch {
fmt.Print(chunk.Content)
if chunk.Usage != nil { /* prompt/completion/total (+ cache fields if reported) */ }
}The API key is your scheduling identity (
Authorization: Bearer …→ a priority on the fair-share proxy), not just auth.TraceIDis correlation only (anX-Trace-Idheader for server logs).
Session.Turn runs chat → tool_calls → dispatch → feed back → repeat. You
supply Tools []llm.ToolDef and a Dispatch:
func dispatch(ctx context.Context, tc llm.ToolCall) (string, error) {
switch tc.Function.Name {
case "get_weather":
// parse tc.Function.Arguments (JSON), do the work, return a string
return result, nil
default:
return fmt.Sprintf("ERROR: unknown tool %q", tc.Function.Name), nil // note: not a Go error
}
}Errors meant for the model (unknown tool, bad args) go into the result
string so the loop stays alive and the model can recover. A returned Go error
aborts the Turn. A terminal tool returns agent.ErrSessionClosed to stop the
loop after its result is persisted (used with tool_choice=required roles).
A client pointed at api.anthropic.com (or with MessagesAPI: true) talks to
/v1/messages. Callers change nothing: messages, tools, options and stream chunks
stay in the OpenAI shape. The body is built as for any provider, translated by
openai-claude-api-shim, and
sent through the same retries; the reply is transcoded back into what readStream
reads. The decision is per client, so a parent on one provider can run a child on
another. The reason is cost: Anthropic's OpenAI-compatible endpoint reads nothing from
the prompt cache, and the shim places breakpoints.
ReasoningEffort becomes output_config.effort; thinking cannot be switched off on
every model, so none is low. A refusal is a stream error. The client needs a key:
Workload Identity Federation went with the SDK. What the shape cannot carry is listed
in the shim's README.
Two different failures, two guards.
A generation that collapses into repetition. Measured: ~30 minutes spent
re-emitting an 85-character fragment, to the server's context limit. Stop
can't catch it (the text isn't known in advance) and a grammar can't (the
output is well-formed) — only something watching the bytes arrive can, so
llm.Client does. It watches content, reasoning and tool-call arguments (the
measured loop was inside arguments), and on a trip it closes the stream, which
disconnects and makes the server abandon the generation.
The caller sees a final chunk carrying the reason; Session.Turn handles it for
you — it trims the redundant copies off the reply before persisting it (so the
retry isn't built from a context primed to continue the loop), tells the model
what it did, and re-prompts. Two consecutive collapses are retried; a third
fails the Turn rather than paying the same cost in installments.
client.Repetition = llm.RepetitionGuard{Off: true} // disable entirely
client.Repetition = llm.RepetitionGuard{MinSpan: 2048} // or just loosen it
session.MaxRepetitionRetries = -1 // never give up (MaxTurns still bounds)Thresholds are set so that ordinary repetition — blank lines, markdown table
rules, indentation, recurring boilerplate — never trips it. A cycle shorter than
MinPeriod (24 bytes) is deliberately not reported at all; ChatOpts.MaxTokens
is the backstop for those.
A turn-level loop: the same call returning the same result, round after
round. Every round is well-formed, so only the repetition across rounds shows
it. MaxRepeatedExchanges (default 5) warns the model at 5 identical
call+result exchanges and stops the Turn at 10.
The result is part of that comparison on purpose: polling is a legitimate reason to repeat a call, so a guard keyed on the call alone would kill a job that is correctly waiting. A poll whose result advances never trips this.
session.MaxRepeatedExchanges = -1 // offA loop with no period: loopdetector. Opt-in. The guards above need identical
bytes back to back or identical rounds. A model can also go round many variations,
in any order, with other output between the copies: one production reply was 291
sentences of which 40 were distinct, each said about 21 times, and no exact guard
saw it. loopdetector.LoopDetector finds those.
Its input is the session plus the stream: Begin(history) takes the request's
messages, Watch(channel, delta) the reply as it arrives, thinking included. There
is no cache; nothing outlives the reply. Its output is a count, and Tolerance says
at what count the stream is cut:
d := loopdetector.New(loopdetector.Config{
Tolerance: 3, // cut at 3 loops. Negative: count, never cut
})
d.Install(client) // replaces client.Repetition for this client's streams
// Or feed it yourself, which is what a proxy does:
s := d.Begin(messages)
if info := s.Watch(llm.ChannelReasoning, delta); info != nil { /* cut */ }
loops, kind := s.Loops()
d.Counters() // totals, for metricsIt finds four kinds: exact (the guard), cycle (the same sentences recurring),
phrase (the same, a little reworded, matched by BM25) and tool (the same call
with the same result; no tool is exempt, so polling for an unchanged result counts,
and a host should wait on an event instead). A window of recent sentences has to be mostly repeats, so a
sentence the model says often is not a loop. History is read from the last message by a
person on, so a person asking again is not a loop, and a loop that goes on only
across their messages is not seen. Host notices ([tag] text) do not end a turn. History is counted but cannot cut: only a reply
that continues the loop is interrupted. ToolTolerance counts tools while text is
cut. A finding carries First and Last, the earliest repeat in view and the one
that tripped it, capped at SampleBytes. Finding() returns it whether or not the
stream was cut. A proxy that found a loop and let the stream pass sends it on the
final event, where it arrives as StreamChunk.Repetition with no StopReason:
nothing was cut, and what it means is the caller's to judge. Random data inside every copy
of a line, and real churn that is new text each time, are not found.
tool_choice=required forces the model to call a tool, not to fill it
correctly. ValidatingDispatcher closes that gap: a call whose arguments fail
validation is rejected with a fix instruction as its tool result — the
session stays active and the model retries. The inner dispatcher never runs on
a bad call.
v := agent.NewSchemaValidator(tools) // required-keys + JSON-kind checks from the tools' own schemas
dispatch := agent.ValidatingDispatcher(realDispatch, v)SchemaValidator is dependency-free and conservative (object shape, required
fields present + non-null, primitive type match). For stricter guarantees, plug
your own agent.Validator (e.g. a full JSON-Schema library) — the interface is
the seam.
The hard counterpart to the (soft) validator: the server constrains token
sampling so the output cannot violate the shape. Two knobs on ChatOpts,
forwarded raw to the endpoint (llama.cpp / corrallm honor them; others ignore):
// GBNF grammar — output must match the grammar.
&llm.ChatOpts{Grammar: `root ::= "red" | "green" | "blue"`}
// JSON mode — output is guaranteed valid JSON.
&llm.ChatOpts{ResponseFormat: map[string]any{"type": "json_object"}}
// or a full json_schema response format:
&llm.ChatOpts{ResponseFormat: map[string]any{
"type": "json_schema",
"json_schema": map[string]any{"name": "weather", "schema": schemaObj},
}}Use constrained decoding when the structure must be guaranteed; use the Validator when you want the model to self-correct with feedback. They compose.
Inject a message or system notification into a session; the next Turn
renders it inline:
sess.Inject(ctx, agent.Entry{Kind: agent.KindNotification, Tag: "deploy",
Content: "Deploy #42 succeeded on prod."})Batching is inherent: ClaimPending marks all pending arrivals shown at
the top of a turn, and the context builder renders every non-subsumed entry —
so several messages that queued between activations reach the model in one
turn, not one turn each.
A tool that can't answer inline (kicked off a job, an approval, a long shell) returns the lift wire shape instead of a result:
{"pending": true, "correlation_id": "job-7", "ttl_s": 30}The dispatcher recognizes it and substitutes a "pending, wrap up" message:
if lr, ok := agent.ParseLiftRequest(result); ok {
result = agent.PendingResult(lr.CorrelationID, tc.ID, lr.TTLSeconds)
// record the pending call host-side, keyed by tc.ID
}The turn ends normally; the session stays active. When the upstream
completes, the host injects the real payload as a KindToolResult Entry keyed
by the same ToolCallID (sess.Inject), and the next Turn reconciles it.
The lift demo runs this end to end: Turn 1 parks on the async tool and the
model acknowledges; the host injects the finished result; Turn 2 resumes and
answers — the turn never blocked on the job. agentkit owns only the wire shape +
wording; storage, the completion endpoint, and deadline GC stay with you.
Lifting is event-driven, never a blocked goroutine.
llm.StreamChunkToSSE formats a StreamChunk as a Server-Sent Events frame —
content deltas as data: {"type":"content","text":"…"}, tool calls as
{"type":"tool_call",…}, plus [DONE] and errors. Relaying a live completion
to a browser is then a ~10-line handler:
ch, _ := client.ChatStream(r.Context(), msgs, tools, nil)
w.Header().Set("Content-Type", "text/event-stream")
for chunk := range ch {
io.WriteString(w, llm.StreamChunkToSSE(chunk))
w.(http.Flusher).Flush()
}The serve demo stands up this endpoint and self-requests it, printing the raw
frames a browser's EventSource would receive.
Notifications are cheap to publish but expensive to consume: a stale in-flight
notice (an MCP-LSP "bad build" for a file that now compiles) is re-validated by
the model every turn until it's cleared. Three primitives, keyed on a notice's
(groupBy, key) partition:
- supersede — a re-emit for the same key replaces the prior unshown notice (newest wins) instead of stacking. The inbox holds ≤1 live notice per key.
- clear — retract the notice when its condition resolves. A tool result or
integration callback carries
agent.ClearRequest({"clear":true,"group_by":"file","key":"main.go"});ParseClearRequestreads it. Zero model cost. - preparer — the
NotificationPreparerhook runs at the top of every turn (after claim, before build) so a resolved notice is gone before it renders.agent.MCPPrepareris the ready-made preparer for the MCP-revalidator convention: for each pending notice, call the integration's masked revalidator tool for its group key; an empty "current truth" means resolved → clear (agent.IsResolvedTruth). Fail-open on a flaky tool.
This is where an active "check back" belongs — an integrator-owned hook that runs whatever it trusts, not daemon-side shell.
Shaper.Build fits history to the window (see concepts.md):
pristine tail → LOD truncation → compaction. LOD is pure render-time (the
stored entry keeps its full content behind an event_id pointer); compaction
folds the oldest prefix into a summary marker via Store.Compact.
To protect the KV cache, reshaping is deferred until the context would cross
BudgetTokens − LODHeadroomTokens (default ~10k), then done in one decisive
pass with runway — not eagerly every turn. A compaction is surfaced
(OnCompaction + TurnResult.Compactions) with its summary + before/after
tokens, and the same turn continues to the reply. Every Turn also reports
TurnResult.Usage — Total billed vs Active window. The compact demo
shows a compaction folding 1150→28 tokens, the summary emitted as a hidden
field, and the total/active tally.
mcpmgr.Manager spawns stdio MCP servers, discovers their tools, and calls
them, with project-scoped (shared) and thread-scoped (per-workspace) instances,
plus 0600 secret-file materialization for servers that need credentials. The
entire integration is two bridges:
mgr := mcpmgr.NewManager()
mgr.StartServer(ctx, mcpmgr.MCPConfig{ID: "everything", Command: "npx",
Args: []string{"-y", "@modelcontextprotocol/server-everything"}})
tools := mgr.GetTools() // discovered MCPTools
defs := mcpToolDefs(tools) // MCPTool → llm.ToolDef (advertise)
dispatch := mcpDispatcher(mgr, tools) // route calls → Manager.CallToolmcpToolDefs copies name/description and drops the MCP InputSchema straight
into ToolDef.Parameters (it's already JSON Schema); mcpDispatcher maps each
tool name to its owning server and calls Manager.CallTool. See
examples/agentkit-demo/mcp.go. mcpmgr is independent of agent — use it
wherever you build tool defs.
The features compose. To get guaranteed typed data out of a model, the
structured demo stacks four of them: tool_choice forces the model to call a
specific tool, the SchemaValidator fix loop gates its arguments,
ForcedTerminalTool makes it the session's only exit, and ErrSessionClosed
ends the loop the instant valid data arrives. Add ChatOpts.Grammar /
ResponseFormat for a hard server-side guarantee instead of the soft fix loop.
Every Session and Shaper takes an optional agent.Tracer (nil = zero
overhead). It captures spans for Turn, streamChat, and Shaper.Build with
attributes (message count, tool calls, token usage). A host adapts its own
tracing onto the two-method interface; examples/agentkit-demo/trace.go is a
~40-line stdout implementation you get with --trace:
┌ agent.Turn
│ ┌ agent.streamChat
│ └ agent.streamChat (16.9s) n_messages=2 n_tool_calls=1 total_tokens=410
└ agent.Turn (16.9s)
The features above are not independent — they coalesce. A lifted tool
result, user messages that queued, and live notifications all converge into one
merged context at the next turn (see
concepts.md → the coalescing turn boundary).
The converge demo runs it end to end: a tool call parks, then a resolved
result + a user message + a notification accumulate while the model is idle, and
the next Turn delivers all three together — the model addresses them in one
reply. This convergence is the point of the event-driven turn model.