building-agents

name	building-agents
description	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 specify first (method: sdd), 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 only imports anthropic → claude-api if your environment has it, since this skill stays multi-provider. Workspace scaffolding → harness. Choosing which 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.

Decision rules (read before writing code)

1. Adapter first — define the LLMProvider Protocol before any provider call.
2. 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.
3. Tools are typed contracts — schema + validation + idempotency key on every side-effecting tool; no catch-all tools.
4. Retrieve, don't stuff — RAG when ground truth lives in data; cite or refuse.
5. Eval before ship — a golden set + regression gate in CI, or it's not production.
6. 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 import Literal, Protocol, runtime_checkable

from pydantic import BaseModel, Field


class Message(BaseModel):
    role: Literal["system", "user", "assistant", "tool"]
    content: str


class ToolSpec(BaseModel):
    name: str
    description: str
    parameters: dict  # JSON Schema for the tool's arguments


class Usage(BaseModel):
    input_tokens: int = 0
    output_tokens: int = 0
    cost_usd: float = 0.0


class CompletionRequest(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:  = 


 ():
    text:  = 
    tool_calls: [] = Field(default_factory=)  
    usage: Usage = Field(default_factory=Usage)
    raw:  |  = 



 ():
    
    
      () -> CompletionResponse: ...


 :
     () -> :
         openai  AsyncOpenAI

        .model, .client = model, AsyncOpenAI()

      () -> CompletionResponse:
        
        
        kwargs:  = {: .model, : [m.model_dump()  m  req.messages],
                        : req.temperature, : req.max_tokens}
         req.tools:
            kwargs[] = [{: , : {: t.name, : t.description, : t.parameters}}  t  req.tools]
         req.response_schema:
            kwargs[] = {: , : {: , : req.response_schema, : }}
        r =  .client.chat.completions.create(**kwargs)
        msg = r.choices[].message
        calls = [{: c., : c.function.name, : c.function.arguments}  c  (msg.tool_calls  [])]
         CompletionResponse(text=msg.content  , tool_calls=calls, raw=r.model_dump(),
            usage=Usage(input_tokens=r.usage.prompt_tokens, output_tokens=r.usage.completion_tokens))


 :
     () -> :
         anthropic  AsyncAnthropic

        .model, .client = model, AsyncAnthropic()

      () -> CompletionResponse:
        
        system = .join(m.content  m  req.messages  m.role == )  
        turns = [{: m.role, : m.content}  m  req.messages  m.role != ]
        kwargs:  = {: .model, : system, : turns, : req.max_tokens, : req.temperature}
         req.tools:
            kwargs[] = [{: t.name, : t.description, : t.parameters}  t  req.tools]
         req.response_schema:  
            kwargs[] = [{: , : , : req.response_schema}]
            kwargs[] = {: , : }
        r =  .client.messages.create(**kwargs)
        text = .join(b.text  b  r.content  b. == )
        calls = [{: b., : b.name, : b.}  b  r.content  b. == ]
         CompletionResponse(text=text, tool_calls=calls, raw=r.model_dump(),
            usage=Usage(input_tokens=r.usage.input_tokens, output_tokens=r.usage.output_tokens))


 () -> LLMProvider:
    
    provider, _, model = (spec  os.environ[]).partition()
     provider == :
         OpenAIAdapter(model)
     provider == :
         AnthropicAdapter(model)
     ValueError()

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.
class Answer(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.
while True:
    resp = await provider.complete(req)
    if not resp.tool_calls:
        break
    for call in resp.tool_calls:
        await run_tool(call)

# GOOD — bounded loop: step cap + per-tool timeout + idempotency key (safe to retry).
for step in range(max_steps):
    resp = await provider.complete(req)
    if not resp.tool_calls:
        break
    for call in resp.tool_calls:
        async with 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 import Callable, Literal

from pydantic import BaseModel, ConfigDict, Field, ValidationError


class CreateInvoiceArgs(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"


class ToolResult(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 + idempotency
    return ToolResult(status="success", summary=f"Created {invoice_id}", data={"id": invoice_id})


TOOLS: dict[str, tuple[[BaseModel], ]] = {
    : (CreateInvoiceArgs, _create_invoice),
}


 () -> ToolResult:
    spec = TOOLS.get(name)
     spec  :
         ToolResult(status=, summary=, next_actions=[])
    args_model, handler = spec
    :
        args = args_model.model_validate(raw_args)  
     ValidationError  e:
         ToolResult(status=, summary=, data={: e.errors()},
                          next_actions=[])
     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 NOT EXISTS vector;
CREATE TABLE IF NOT EXISTS docs (
    id        bigserial PRIMARY KEY,
    content   text NOT NULL,
    embedding vector(1536) NOT NULL,
    meta      jsonb NOT NULL DEFAULT '{}'
);
CREATE INDEX IF NOT EXISTS docs_embedding_hnsw
    ON docs USING hnsw (embedding vector_cosine_ops);

async def embed(texts: list[str]) -> list[list[float]]:
    # Same provider interface as completions; impl in references/tools-and-rag.md.
    return await provider.embed(texts)  # returns one 1536-d vector per text


async def retrieve(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]


async def answer(query: str) -> str:
    chunks = await retrieve(query)
    if not chunks:                                   # refuse rather than hallucinate
        return "I don't have grounded information to answer that."
    context = "\n".join(  c  chunks)
    req = CompletionRequest(
        model=settings.model_id,
        messages=[Message(role=, content=),
                  Message(role=, content=)],
    )
     ( provider.complete(req)).text

Evals & cost gates (the production line)

import json
import statistics
import sys
import time


async def run_eval(golden_path: str, graders: list, thresholds: dict[str, float]) -> None:
    cases = [json.loads(line) for line in open(golden_path)]  # {"input","expected","meta"}
    results = []
    for case in 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[]  r  results], n=)[-],
        : (r[]  r  results) / n,
    }
    failed = [k  k, lo  thresholds.items()  metrics[k] < lo]
    (json.dumps(metrics, indent=))
    sys.exit(  failed  )  

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.

Observability (OTel GenAI, vendor-neutral)

from opentelemetry import trace

tracer = trace.get_tracer("agent")


async def traced_complete(provider: LLMProvider, req: CompletionRequest) -> CompletionResponse:
    with tracer.start_as_current_span("chat") as span:
        span.set_attribute("gen_ai.system", settings.llm.split(":")[0])
        span.set_attribute("gen_ai.request.model", req.model)
        resp = await provider.complete(req)
        span.set_attributes({"gen_ai.usage.input_tokens": resp.usage.input_tokens,
                             "gen_ai.usage.output_tokens": resp.usage.output_tokens,
                             "gen_ai.usage.cost_usd": resp.usage.cost_usd})
        return resp
# Langfuse / Phoenix / Braintrust are swappable OTLP backends: emit spans, swap the exporter.
# span-per-tool, trace-id propagation, exporters -> references/evals-and-observability.md

MCP: when and the smallest server

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()
def create_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}")
def read_invoice(invoice_id: str) -> str:
    """Read-only invoice lookup by id."""
    return f"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.

See also

• Stacks the examples target: fastapi, nextjs, go, postgresdb, flutter. Harden with secure-coding, ship with deployment.
• 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.