Use when building or restructuring an LLM agent — provider adapter, tool calling, structured output, RAG, agent loop, eval gate, cost routing, tracing, MCP server — model-agnostic across OpenAI/Anthropic/Gemini/OSS so a model swap is a config change. NOT vector-store SQL alone (that is `postgresdb`) or service deployment (that is `deployment`).
Use when building or restructuring an LLM agent — provider adapter, tool calling, structured output, RAG, agent loop, eval gate, cost routing, tracing, MCP server — model-agnostic across OpenAI/Anthropic/Gemini/OSS so a model swap is a config change. NOT vector-store SQL alone (that is `postgresdb`) or service deployment (that is `deployment`).
tags
["agents","llm","mcp","rag","evals","ai"]
recommends
["secure-coding","deployment"]
origin
risco
Building production LLM agents (model-agnostic)
A thin provider adapter, a disciplined agent loop, schema-validated tools, provider-neutral RAG, eval gates, OTel tracing, and optionally an MCP server — so swapping OpenAI ↔ Anthropic ↔ Gemini ↔ OSS is a config change, not a rewrite.
The one rule
Program against a capability interface, never a vendor SDK. Vendor specifics (model id, tool-schema shape, JSON mode, caching, token limits) live behind one adapter resolved from config. Model names and prices rot — if one appears in business logic it's a bug, and re-verify the dated tables before quoting a number.
Hand off instead when: a new non-trivial feature has no approved spec + plan under 02-DOCS/wiki/sdd/ → stop and run first (method: ), which routes back here once the plan is approved; one-line/low-risk changes go straight through. Anthropic-SDK internals (caching, thinking, batch) in a file that imports → if your environment has it, since this skill stays multi-provider. Workspace scaffolding → . Choosing coding agent to use → agent-eval territory. Pure prompt-wording tuning with no architecture change → prompt engineering, not this. A one-shot throwaway prompt, or no retrieval/tools/loop/evals at all → you don't need an agent; call the SDK directly and say so.
Adapter first — define the LLMProvider Protocol before any provider call.
Smallest loop that works — single-agent before multi-agent; ReAct only when the path is uncertain; plan-execute when steps are knowable. Multi-agent means orchestrator-worker with a semaphore-bounded parallel fan-out, never a free-for-all.
Tools are typed contracts — schema + validation + idempotency key on every side-effecting tool; no catch-all tools.
Retrieve, don't stuff — RAG when ground truth lives in data; cite or refuse.
Eval before ship — a golden set + regression gate in CI, or it's not production.
Cheapest model that passes the eval — route/cascade up, never default to flagship.
The provider adapter (the heart of the skill)
The one payload to internalize. Python 3.12+, Pydantic v2, async so it composes directly with the bounded loop (and orchestrator-worker fan-out) in references/agent-loops-and-harness.md. Structured output is the quirk that differs most per vendor: strict JSON Schema (OpenAI), tool-forcing (Anthropic), response_json_schema (Gemini). Streaming, the Gemini and OSS/litellm adapters, tool-result plumbing, and a route() registry live in references/provider-abstraction.md — this excerpt is the load-bearing core, not the whole interface.
from __future__ import annotations
import os
from typing importLiteral, Protocol, runtime_checkable
from pydantic import BaseModel, Field
classMessage(BaseModel):
role: Literal["system", "user", "assistant", "tool"]
content: strclassToolSpec(BaseModel):
name: str
description: str
parameters: dict# JSON Schema for the tool's argumentsclassUsage(BaseModel):
input_tokens: int = 0
output_tokens: int = 0
cost_usd: float = 0.0classCompletionRequest(BaseModel):
model: str# resolved from config, e.g. "claude-sonnet-4-6" — never literal in logic
messages: list[Message]
tools: list[ToolSpec] = Field(default_factory=list)
response_schema: dict | None = None# JSON Schema -> structured output
temperature: float = 0.0
max_tokens: int = 1024classCompletionResponse(BaseModel):
text: str = ""
tool_calls: list[dict] = Field(default_factory=list) # [{id, name, arguments}]
usage: Usage = Field(default_factory=Usage)
raw: dict | None = None@runtime_checkableclassLLMProvider(Protocol):
# Async so it drives the async agent loop directly. The full interface in# references/provider-abstraction.md adds stream() and embed().asyncdefcomplete(self, req: CompletionRequest) -> CompletionResponse: ...
classOpenAIAdapter:
def__init__(self, model: str) -> None:
from openai import AsyncOpenAI
self.model, self.client = model, AsyncOpenAI()
asyncdefcomplete(self, req: CompletionRequest) -> CompletionResponse:
# Chat Completions shape (universal, still current); references/provider-abstraction.md# gives the preferred Responses-API adapter. system stays a `system` role message here.
kwargs: dict = {"model": self.model, "messages": [m.model_dump() for m in req.messages],
"temperature": req.temperature, "max_tokens": req.max_tokens}
if req.tools:
kwargs["tools"] = [{"type": "function", "function": {"name": t.name, "description": t.description, "parameters": t.parameters}} for t in req.tools]
if req.response_schema:
kwargs["response_format"] = {"type": "json_schema", "json_schema": {"name": "out", "schema": req.response_schema, "strict": True}}
r = awaitself.client.chat.completions.create(**kwargs)
msg = r.choices[0].message
calls = [{"id": c.id, "name": c.function.name, "arguments": c.function.arguments} for c in (msg.tool_calls or [])]
return CompletionResponse(text=msg.content or"", tool_calls=calls, raw=r.model_dump(),
usage=Usage(input_tokens=r.usage.prompt_tokens, output_tokens=r.usage.completion_tokens))
classAnthropicAdapter:
def__init__(self, model: str) -> None:
from anthropic import AsyncAnthropic
self.model, self.client = model, AsyncAnthropic()
asyncdefcomplete(self, req: CompletionRequest) -> CompletionResponse:
# QUIRKS: system is a top-level param (not a message); tools use input_schema (not function).
system = "\n".join(m.content for m in req.messages if m.role == "system") orNone
turns = [{"role": m.role, "content": m.content} for m in req.messages if m.role != "system"]
kwargs: dict = {"model": self.model, "system": system, "messages": turns, "max_tokens": req.max_tokens, "temperature": req.temperature}
if req.tools:
kwargs["tools"] = [{"name": t.name, "description": t.description, "input_schema": t.parameters} for t in req.tools]
if req.response_schema: # structured output via tool-forcing
kwargs["tools"] = [{"name": "out", "description": "Emit the result", "input_schema": req.response_schema}]
kwargs["tool_choice"] = {"type": "tool", "name": "out"}
r = awaitself.client.messages.create(**kwargs)
text = "".join(b.text for b in r.content if b.type == "text")
calls = [{"id": b.id, "name": b.name, "arguments": b.input} for b in r.content if b.type == "tool_use"]
return CompletionResponse(text=text, tool_calls=calls, raw=r.model_dump(),
usage=Usage(input_tokens=r.usage.input_tokens, output_tokens=r.usage.output_tokens))
defget_provider(spec: str | None = None) -> LLMProvider:
"""Parse 'provider:model' (default from env LLM) into a concrete adapter."""
provider, _, model = (spec or os.environ["LLM"]).partition(":")
if provider == "openai":
return OpenAIAdapter(model)
if provider == "anthropic":
return AnthropicAdapter(model)
raise ValueError(f"unknown provider: {provider!r}")
# Gemini + OSS/litellm adapters, streaming, tool-result plumbing, and route() registry# -> references/provider-abstraction.md
Good vs Bad
Call-sites use the adapter and never name a model: provider = get_provider(settings.llm) (e.g. "anthropic:claude-sonnet-4-6"), then await provider.complete(req). The two failures that survive that discipline:
# BAD — parse-and-pray; wrong shape fails silently at 3am.
raw = (await provider.complete(req)).text
try:
data = json.loads(raw)
except json.JSONDecodeError:
data = {} # the bug is now invisible
# GOOD — strict structured output + schema validation that fails loudly on drift.classAnswer(BaseModel):
sentiment: Literal["pos", "neg", "neu"]
score: float
req.response_schema = Answer.model_json_schema()
ans = Answer.model_validate_json((await provider.complete(req)).text)
# BAD — unbounded loop; no cap/timeout/idempotency. Burns budget, repeats side effects, wedges.whileTrue:
resp = await provider.complete(req)
ifnot resp.tool_calls:
breakfor call in resp.tool_calls:
await run_tool(call)
# GOOD — bounded loop: step cap + per-tool timeout + idempotency key (safe to retry).for step inrange(max_steps):
resp = await provider.complete(req)
ifnot resp.tool_calls:
breakfor call in resp.tool_calls:
asyncwith asyncio.timeout(tool_timeout_s):
await run_tool(call, idempotency_key=call["id"])
# full loop, budgets, recovery -> references/agent-loops-and-harness.md
Tools & structured output (minimum viable)
from typing importCallable, Literalfrom pydantic import BaseModel, ConfigDict, Field, ValidationError
classCreateInvoiceArgs(BaseModel):
model_config = ConfigDict(extra="forbid") # reject unknown keys from the model
customer_id: str = Field(min_length=1)
amount_cents: int = Field(gt=0)
currency: Literal["EUR", "USD"] = "EUR"classToolResult(BaseModel):
status: Literal["success", "warning", "error"]
summary: str
data: dict | None = None
next_actions: list[str] = Field(default_factory=list)
def_create_invoice(args: CreateInvoiceArgs) -> ToolResult:
invoice_id = f"inv_{args.customer_id}_{args.amount_cents}"# real impl: DB insert + idempotencyreturn ToolResult(status="success", summary=f"Created {invoice_id}", data={"id": invoice_id})
TOOLS: dict[str, tuple[type[BaseModel], Callable]] = {
"create_invoice": (CreateInvoiceArgs, _create_invoice),
}
defdispatch(name: str, raw_args: dict) -> ToolResult:
spec = TOOLS.get(name)
if spec isNone:
return ToolResult(status="error", summary=f"unknown tool {name!r}", next_actions=["pick a registered tool"])
args_model, handler = spec
try:
args = args_model.model_validate(raw_args) # validate BEFORE side effectsexcept ValidationError as e:
return ToolResult(status="error", summary="invalid args", data={"errors": e.errors()},
next_actions=["fix the arguments and retry"])
return handler(args)
Schema design, sandboxing, idempotency, DI-scoped DB sessions, plus the RAG internals below — chunking, hybrid RRF, rerank, the citation grader, memory — are in references/tools-and-rag.md.
RAG in 30 lines (provider-agnostic embeddings)
CREATE EXTENSION IF NOTEXISTS vector;
CREATE TABLE IF NOTEXISTS docs (
id bigserial PRIMARY KEY,
content text NOT NULL,
embedding vector(1536) NOT NULL,
meta jsonb NOT NULLDEFAULT'{}'
);
CREATE INDEX IF NOTEXISTS docs_embedding_hnsw
ON docs USING hnsw (embedding vector_cosine_ops);
asyncdefembed(texts: list[str]) -> list[list[float]]:
# Same provider interface as completions; impl in references/tools-and-rag.md.returnawait provider.embed(texts) # returns one 1536-d vector per textasyncdefretrieve(query: str, k: int = 5, min_sim: float = 0.25) -> list[dict]:
[q] = await embed([query])
rows = await db.fetch( # cosine distance <=>; similarity = 1 - distance"SELECT id, content, 1 - (embedding <=> $1) AS sim ""FROM docs ORDER BY embedding <=> $1 LIMIT $2",
q, k,
)
return [dict(r) for r in rows if r["sim"] >= min_sim]
asyncdefanswer(query: str) -> str:
chunks = await retrieve(query)
ifnot chunks: # refuse rather than hallucinatereturn"I don't have grounded information to answer that."
context = "\n".join(f"[{c['id']}] {c['content']}"for c in chunks)
req = CompletionRequest(
model=settings.model_id,
messages=[Message(role="system", content="Answer ONLY from context; cite chunk ids like [12]."),
Message(role="user", content=f"{context}\n\nQ: {query}")],
)
return (await provider.complete(req)).text
Evals & cost gates (the production line)
import json
import statistics
import sys
import time
asyncdefrun_eval(golden_path: str, graders: list, thresholds: dict[str, float]) -> None:
cases = [json.loads(line) for line inopen(golden_path)] # {"input","expected","meta"}
results = []
forcasein cases:
t0 = time.perf_counter()
out = await provider.complete(CompletionRequest(model=settings.model_id,
messages=[Message(role="user", content=case["input"])]))
scores = {g.name: g.grade(case, out) for g in graders} # exact / schema / LLM-judge
results.append({"scores": scores, "cost": out.usage.cost_usd,
"ms": (time.perf_counter() - t0) * 1000})
n = len(results)
metrics = {
"accuracy": sum(r["scores"]["exact"] for r in results) / n,
"faithfulness": sum(r["scores"]["judge"] for r in results) / n,
"p95_latency_ms": statistics.quantiles([r["ms"] for r in results], n=20)[-1],
"cost_per_task": sum(r["cost"] for r in results) / n,
}
failed = [k for k, lo in thresholds.items() if metrics[k] < lo]
print(json.dumps(metrics, indent=2))
sys.exit(1if failed else0) # CI gate: non-zero blocks the merge
Routing cascade in one line: route(task) → cheapest model whose eval passes; escalate only on a failed self-check.
Full runner, judge, CI gate, caching, batching and budgets →
references/evals-and-observability.md.
Native tools when the agent and tools share a process/repo. MCP when tools must be reused across clients/teams or run out-of-process — accept the MCP cost (schema tokens, transport, ops) in exchange for reuse. TypeScript server, transports, HTTP+auth and testing are in references/mcp-servers.md.
from fastmcp import FastMCP # standalone fastmcp 2.x
mcp = FastMCP("invoices")
@mcp.tool()defcreate_invoice(customer_id: str, amount_cents: int, currency: str = "EUR") -> dict:
"""Create an invoice. amount_cents must be > 0."""if amount_cents <= 0:
raise ValueError("amount_cents must be positive")
return {"id": f"inv_{customer_id}_{amount_cents}", "currency": currency}
@mcp.resource("invoice://{invoice_id}")defread_invoice(invoice_id: str) -> str:
"""Read-only invoice lookup by id."""returnf"Invoice {invoice_id}: status=open"if __name__ == "__main__":
mcp.run() # stdio transport# (MCP spec 2025-11-25; stateless-core RC 2026-07-28; verify before quoting)
Anti-patterns
Anti-pattern
Reality
"I'll just call the OpenAI SDK directly, we'll never switch"
The adapter is ~40 lines; retrofitting it across 30 call-sites later is a rewrite. Adapter first.
"JSON output is usually valid, I'll parse it"
"Usually" = pages at 3am. Use strict structured output + schema validation.
"The agent loop works, I don't need a step cap"
Unbounded loops burn budget and wedge on errors. Cap steps, timeouts, and budget.
"One mega-tool that takes a freeform command is flexible"
It's unobservable and unsafe. Narrow typed tools with idempotency keys.
"We can eval by eyeballing outputs"
Vibes don't gate CI. Golden set + graders + threshold or it's not production.
"Default everything to the flagship model, it's smartest"
5–20× cost for no measured gain. Route to the cheapest model that passes the eval.
"Stuff the whole doc in the prompt instead of RAG"
Blows context + cost and still hallucinates. Retrieve + cite + refuse.
"Retry on every exception"
Retrying a 400/401 wastes budget. Retry only transient (429/5xx/timeout) with backoff+jitter.
"Hardcode the model name, it's fine"
Names rot (Opus 4.7 → 4.8 in weeks). Resolve from config/registry.
"MCP for everything"
In-process native tools are simpler and faster when reuse isn't needed. MCP only for cross-client reuse.
"Tool results just return the raw API blob"
Give the model status/summary/next_actions; raw blobs waste context and stall recovery.
"Prompt caching is Anthropic-only so skip caching"
Each provider has its own caching/dedup; abstract it behind the adapter, don't skip it.
verify.sh
scripts/verify.sh lints example agent code and dry-runs the eval smoke test in the user's project — not in this skill repo. It detects each tool (ruff, mypy, tsc/node, go, the eval entrypoint, markdownlint) and skips any that are missing with a yellow WARN; a missing tool never fails the run. Invoke it with bash scripts/verify.sh from the project root. Exit 0 means clean (or only skips); a non-zero exit means a real lint/typecheck/vet/eval failure.
Project grounding
In a project with a 02-DOCS/ layer (harness), read 02-DOCS/wiki/stack/agents.md first on every use and stay consistent with it. If it is missing or stale, write this project's real choices there — provider(s) and model routing, where the adapter lives, tool/RAG conventions, eval gates, observability backend — as a type: stack article per the harness wiki-article-template.md, index it in 02-DOCS/wiki/index.md, and bump its timestamp in the same change as any convention change. No 02-DOCS/? Skip silently — technical conventions here are recorded, not gated; never block the task on this.
External (no sibling here; use if your environment provides them): claude-api for Anthropic-SDK-only tuning, deep-research for the research-harness fan-out / verify pattern.
