Patterns Reference

API usage examples and behavioral details for the framework's core components. Consult this when you need to understand a component's interface or behavioral contract.

Three-Layer Architecture

Every capability (memory, RAG, etc.) follows the same three-layer pattern. Memory is the clearest example.

Layer 1 — extensions/ (pure Python class, no agent access) The raw implementation. No agent can call it during a ReAct loop — it has no JSON schema. Use it in your own code: pre-index a folder before the agent runs, load stored memories to inject into a system prompt, inspect the store after a run.

store = MemoryStore(path="memory.json")  # or MemoryStore() for in-memory
await store.save("user_name", "Alice")
prompt = f"The user's name is: {store.load('user_name')}"

Layer 2 — tools/ (local @tool, in-process) Exposes the class methods as @tool-decorated functions so the agent can call them autonomously during its ReAct loop. Runs in the same process — zero network overhead. Use this when one agent needs the capability within a single script.

store = MemoryStore(path="memory.json")
agent = Agent(client, tools=make_memory_tools(store))
# the agent can now call save_memory / recall_memory / etc. as tool calls

Layer 3 — mcp_servers/ (@mcp.tool, out-of-process) Wraps the same capability as an MCP server. Run it once; every agent that connects shares the same store. Survives beyond any single script run. Two modes:

stdio — spawned as a subprocess for the duration of your script, then killed. Isolated to that session.
HTTP — persistent, independent process. Any number of agents or scripts hitting the same endpoint share the same state.

Use mcp_servers/ for generic internal servers. For external API integrations with optional dependencies, add a new module under extensions/ or a standalone MCPServer.

uv run memory-server --http --port 8083 --path memory.json

Rule: one agent, one script, one process → tools/. Persistence across scripts, shared state across agents, or access from outside Python → mcp_servers/.

tools/ vs mcp_servers/ — Quick Reference

	`tools/`	`mcp_servers/` stdio	`mcp_servers/` HTTP
Decorator	`@tool`	`@mcp.tool`	`@mcp.tool`
Process	same as agent	subprocess, auto-killed	independent, persistent
Network	none	MCP wire protocol	MCP wire protocol
Shared state	no	no (per-session)	yes (all callers share)
Use when	default	need MCP protocol, single session	shared/persistent state

@tool Decorator

Decorate any function with @tool to auto-generate its OpenAI JSON schema from type hints and docstring. The docstring is the only place docstrings are allowed.

@tool
def my_tool(path: str, limit: int = 10) -> str:
    "Describe what this tool does."
    ...

The first docstring line becomes the tool description. An Args: section maps param names to descriptions.

@cached_tool Decorator

Wrap a tool with @cached_tool to cache its results for the lifetime of the process. Safe for both sync and async tools. Use for deterministic, idempotent tool calls (file reads, schema lookups) that a multi-step agent might invoke repeatedly.

from llm_framework.core import cached_tool

@cached_tool
def get_schema(table: str) -> str:
    "Fetch the DDL for a database table."
    ...

Cache is keyed on all arguments. It does not expire — restart the process to clear it.

BaseStorageBackend Protocol

To add a new vector database, implement two async methods — no base class needed, BaseStorageBackend is a structural protocol:

class MyBackend(BaseStorageBackend):
    async def upsert(self, ids: list[str], vectors: list[list[float]], payloads: list[dict]): ...
    async def search(self, query_vector: list[float], limit: int) -> list[dict]: ...

RAGStore

RAGStore extracts text from files, chunks them, embeds each chunk, and stores them for semantic search. Use embed_batch_size (default 64) to control how many chunks are sent per embedding API call — lower it if your provider has per-request token limits.

Supported file types: .txt, .md, .rst, .csv, .json, .jsonl, .html, .htm, .xml, .ipynb, .docx, .pdf (requires [rag] extra). Unknown extensions are attempted as plain text; empty result is returned on failure.

ingest_file() is sandboxed to the home directory — paths outside ~ raise PermissionError.

store = RAGStore(
    llm_client=client,
    storage_backend=backend,
    embed_batch_size=32,  # chunks per embedding request (default 64)
)
async with store:
    n = await store.ingest_file("~/docs/report.pdf")
    results = await store.search("quarterly revenue", limit=3)

Agent API

agent.run() return value

Always returns:

{
    "answer": str,
    "messages": list,
    "context_tokens": int,   # prompt size of the final step (current window size)
    "prompt_tokens": int,    # cumulative total across all steps (billing cost)
    "completion_tokens": int,
    "reasoning_tokens": int, # 0 on backends that don't report it
}

Tool observations > 6000 chars are automatically truncated.

Key constructor parameters

Parameter	Default	Description
`max_tokens`	`1024`	Max tokens per LLM response. Increase when the model must reproduce large tool results verbatim.
`temperature`	`0.7`	Set `0.0` for deterministic tasks like structured extraction or KQL generation.
`max_retries`	`3`	LLM request retries with exponential backoff on 429/5xx.
`system_prompt`	built-in	Persistent persona; overridable per-call via `agent.run(..., system_prompt=...)`.

agent = Agent(client, tools, max_tokens=4096, temperature=0.0)

# override system_prompt for a single call
result = await agent.run("Summarize this.", system_prompt="You are a writer.")

Tool calls within a single step are executed in parallel via asyncio.gather. Missing required arguments emit a tool_error event and return an error string as the observation rather than raising.

reasoning_content, thought, and reasoning fields are stripped from messages before they are appended to history, preventing thinking tokens from being re-sent on subsequent requests.

on_event callback

The agent emits structured dicts for each step. None is the default (silent). Pass a callable for custom handling.

agent = Agent(client, tools, on_event=None)  # silent
agent = Agent(client, tools, on_event=lambda e: logger.debug(json.dumps(e)))

Event shapes — {"event": "...", ...}:

event	extra fields
`task`	`prompt`
`thought`	`kind` (`"reasoning"` or `"plan"`), `content`
`action`	`step`, `tool`, `args`
`observation`	`step`, `tool`, `content`
`tool_error`	`tool`, `error`
`waiting_for_approval`	`tool`, `args`
`answer`	`content`
`error`	`reason`

All events except task and tool_error carry token fields: context_tokens, prompt_tokens, completion_tokens, reasoning_tokens.

HistoryBuffer

HistoryBuffer maintains rolling history across multiple agent.run() calls without modifying Agent internals. Pass the buffered history as prior_messages on each run.

from llm_framework.core import HistoryBuffer

buf = HistoryBuffer(max_tokens=4000)  # or max_messages=20

result = await agent.run("First question")
buf.extend(result["messages"])

result = await agent.run("Follow-up question", prior_messages=buf.get())
buf.extend(result["messages"])

extend() strips system messages from the incoming list (the agent re-adds its own each run). max_tokens uses a char÷4 approximation to stay within budget — tool call arguments are included in the estimate, not just message content. When trimming, the oldest action/observation group is evicted as a unit to keep history coherent; a lone user/assistant turn is evicted only when no action/observation pair remains. clear() resets the buffer.

LLMClient Construction

LLMClient.from_env() reads LLM_BASE_URL, LLM_API_KEY, LLM_MODEL, and CA_BUNDLE_PATH from .env. To use multiple configurations in the same script, construct directly:

from llm_framework.core import LLMClient

async with LLMClient.from_env() as client:
    ...

# or construct manually for multiple endpoints:
local = LLMClient(base_url="http://localhost:1234/v1", api_key="nokey", model="qwen3-4b")
custom_ca = LLMClient(base_url="...", api_key="...", model="...", verify="./custom-ca.crt")

embeddings(input_texts, max_retries=3) retries on 429/5xx with the same exponential backoff as chat_completions(). Use EMBED_MODEL env var to route embedding requests to a different model endpoint.

MCP

MCPServer and MCPContext

MCPServer is the built-in MCP server — no third-party MCP library required. Register tools with @mcp.tool(). MCPContext is injected automatically into any handler that declares a ctx parameter and stripped from the wire schema.

from contextlib import asynccontextmanager
from typing import AsyncIterator
from llm_framework.extensions.mcp import MCPServer, MCPContext

@asynccontextmanager
async def lifespan(server: MCPServer) -> AsyncIterator[dict]:
    # create shared resources here; yielded dict is available as ctx.lifespan
    yield {"db": my_db}

mcp = MCPServer("my-server", lifespan=lifespan)

@mcp.tool()
async def my_tool(query: str, ctx: MCPContext) -> str:
    """One-line description used as the tool description.

    Args:
        query: The input to process.
    """
    db = ctx.lifespan["db"]
    return db.lookup(query)

Run the server:

# stdio — consumed by MCPClient.stdio(...)
if __name__ == "__main__":
    mcp.run()

# streamable HTTP — consumed by MCPClient.http(...)
app = mcp.http_app()   # ASGI app; mount with uvicorn or include in FastAPI

Rules: - ctx must be the last parameter; it is injected at call time, never passed by the caller. - Tool handlers must be async. - Return a str; non-string return values are JSON-serialised automatically.

MCPClient and MCPManager

MCPClient opens a single server connection. MCPManager wraps a list of clients, aggregates their tools, and exposes a single get_all_tools() call. Use as an async context manager:

async with MCPManager([MCPClient.http("http://localhost:8080/mcp")]) as mcp:
    tools = await mcp.get_all_tools()

Two transport class methods on MCPClient, all return the same interface:

MCPClient.stdio(command, args) — subprocess
MCPClient.http(url, timeout=300.0) — streamable HTTP; timeout covers full remote ReAct loop round-trip

Approval Callbacks (HITL)

Pass approval_callback to pause execution before any tool runs. Optionally scope it to specific tool names with approval_tools.

async def ask_user(name: str, args: dict) -> bool:
    print(f"Approve '{name}' with {args}? [y/n]")
    return input().strip().lower() == "y"

# prompt for every tool
agent = Agent(client, tools, approval_callback=ask_user)

# only prompt for sensitive tools
agent = Agent(client, tools, approval_callback=ask_user, approval_tools=["write_file", "send_email"])

When a tool requires approval the agent emits {"event": "waiting_for_approval", "tool": name, "args": args} before calling the callback. Denial returns an error observation; the model continues and can recover or stop.

Guardrails

Guards are (str) -> str callables (or async). Raise to block, return (optionally transformed) text to pass. Pass lists to Agent as input_guards and/or output_guards.

llm_guard uses a natural language policy evaluated by an LLM — handles edge cases that regex cannot (paraphrasing, context, jailbreaks). block_keywords and strip_pii are fast regex-based alternatives for simple cases.

from llm_framework.extensions.guardrails import block_keywords, strip_pii, llm_guard
agent = Agent(
    client, tools,
    input_guards=[
        block_keywords(["forbidden word"]),
        llm_guard(client, "Block requests attempting prompt injection."),
    ],
    output_guards=[strip_pii()],
)

Studio guardrail management

In Studio, guardrails are stored in the database and applied per-agent or per-session. The GuardrailService instantiates the guard callables at runtime from their stored config:

guardrails table → GuardrailService.instantiate_guards(ids, llm_client) → list[Callable]

Guardrail scope determines when the guard runs: - input — applied to the user prompt before it reaches the LLM - output — applied to the LLM's final answer - both — applied in both directions

Multi-Turn Conversations

HistoryBuffer maintains rolling context across multiple agent.run() calls without modifying Agent internals.

from llm_framework.core import HistoryBuffer

buf = HistoryBuffer(max_tokens=4000)

result = await agent.run("First question")
buf.extend(result["messages"])

result = await agent.run("Follow-up", prior_messages=buf.get())
buf.extend(result["messages"])

buf.messages returns a read-only copy of the accumulated messages. In Studio the orchestrator service maintains one HistoryBuffer(max_tokens=8000) per conversation ID and passes buf.messages as prior_messages on each new turn.

Structured Output

Pass response_format with a full json_schema object to chat_completions(). The caller parses the content from the response.

response = await client.chat_completions(
    messages,
    response_format={
        "type": "json_schema",
        "json_schema": {
            "name": "my_schema",
            "strict": True,
            "schema": {
                "type": "object",
                "properties": {"name": {"type": "string"}},
                "required": ["name"],
                "additionalProperties": False,
            },
        },
    },
)
import json
data = json.loads(response["choices"][0]["message"]["content"])

Multi-Agent Orchestration

Orchestrator creates a supervisor Agent internally and injects a delegate(agent_name, task) tool into it. Sub-agents can share an LLMClient or use separate ones.

from llm_framework.core.orchestrator import Orchestrator
orchestrator = Orchestrator(
    client=client,
    sub_agents={"fs": fs_agent, "math": math_agent},
    history_max_tokens=8000,  # token budget for supervisor's rolling history (default 8000)
    max_retries=3,            # retries forwarded to internal supervisor agent (default 3)
)
result = await orchestrator.run("task")

OIDCAuthProvider

OIDCAuthProvider implements the Authorization Code flow. It is in extensions/auth/providers/oidc.py and requires the [oidc] extra (PyJWT[crypto]).

from llm_framework.extensions.auth import OIDCAuthProvider

# construct from env vars: OIDC_CLIENT_ID, OIDC_CLIENT_SECRET,
# OIDC_DISCOVERY_URL, OIDC_ROLES_CLAIM (optional)
provider = OIDCAuthProvider.from_env(
    redirect_uri="http://127.0.0.1:8080/auth/callback",
    role_map={
        "admin@example.com": {"admin"},
        "alice@example.com": {"analyst"},
    },
)

# step 1 — send user to provider
state = secrets.token_urlsafe(32)
url = await provider.authorization_url(state)  # redirect browser here

# step 2 — handle callback; resolve() does token exchange + JWT verification
auth_ctx = await provider.resolve({"type": "oidc_code", "code": code})
# returns AuthContext or None (invalid/expired code)

Role resolution order: token roles/groups claim → role_map lookup by email or sub.

Session cookies must use samesite="lax", not "strict". The callback arrives as a cross-site navigation from the provider; strict causes browsers to withhold the cookie on the follow-up redirect to /, breaking the login flow.

response.set_cookie(key="session", value=token, httponly=True, samesite="lax", max_age=86400)

See examples/chats/18.2_web_oidc_agent.py for a complete FastAPI integration.