agent-gantry

Imported from CodeHalwell/Agent-Gantry (agent_gantry/skills/agent-gantry/SKILL.md). Install upstream with npx skills add CodeHalwell/Agent-Gantry --skill agent-gantry. Copyright stays with the author.

Agent-Gantry

Universal semantic tool router for LLM agents. Register tools once with @gantry.register, sync them into a vector store, then have the router surface only the top-K relevant tools per query — typically cutting prompt token spend by ~80% versus listing every tool.

This skill is the canonical reference for using the library. Read the section that matches the user's framework before writing code. When a user reports "the LLM doesn't see tool X" or "my surface is empty", jump straight to Debugging routing below — that's the most common failure mode and the library ships first-class introspection for it.

When to use which integration path

Native adapters vs. fetch_framework_tools. The per-framework <Framework>Adapter(gantry).select(...) methods return the framework's native tool objects (e.g. a LangChain StructuredTool, a CrewAI BaseTool), with execution routed through gantry.execute (retries, timeouts, circuit breakers, security policy). Prefer these. fetch_framework_tools still exists but only emits OpenAI-shape JSON schemas and supports a smaller name set — see its note.

Core concepts (read first)

1. Tools are functions decorated with @gantry.register. Type hints + docstring become the schema and the embedding text.

   examples=[...] is the single highest-value thing to add. It is the text the router embeds and a selector reads. On our own benchmark it moved the default embedder from 1/5 to 5/5 correct — further than switching to a larger embedding model did. If a user reports poor retrieval, check for missing examples before suggesting a different embedder. Write the phrases a user would actually type, not a paraphrase of the description:

   @gantry.register(
       tags=["weather"],
       examples=["what's the weather in London", "is it raining in Leeds"],
   )
   def get_weather(location: str) -> str:
       """Get the current weather in a given location."""

2. await gantry.sync() embeds every tool. Fingerprint-based change detection means only modified tools re-embed on subsequent calls. With paid embedders, wrap the embedder in CachedEmbedder to persist across cold starts.

3. gantry.retrieve(...) returns a RetrievalResult. That's the universal API. The high-level integrations (AgentFrameworkAdapter/GantryContextProvider, GantryToolBridge, the <Framework>Adapter native adapters, the per-SDK LLM adapters, ToolRefresher, @with_semantic_tools) are thin wrappers around it that emit framework-specific shapes.

   Every native adapter exposes <Framework>Adapter(gantry).select(query, *, limit=3, **select_kwargs) (async) and shares the same selection knobs (score_threshold, namespaces, tools_already_used). Import the class from the clean per-framework namespace — from agent_gantry.langchain import LangChainAdapter. Importing agent_gantry never pulls in any third-party framework; the import is lazy and raises a pip install ... hint only when you call an adapter method.

4. score_threshold is opt-in filtering. Default is 0.0 (no filtering). Use score_threshold="relative:0.8" for length-robust filtering — that keeps anything within 80% of the top score. Fixed absolute cutoffs degrade badly on long, instruction-style queries because the embedding gets diluted.

5. query_strategy="per_call" re-runs retrieval every chat-completion round. This is the right choice for multi-step agents. Pair it with provider.attach_to(agent) or middleware=[provider.as_chat_middleware()]. Without the middleware, per_call silently degrades to per_run and the provider will warn you once.

6. Observability ships built-in — don't hand-roll it. provider.trace() (or provider.attach_to(agent, trace=True)) prints a readable per-round line of what was called and surfaced; provider.selections keeps the per-round retrieval history; gantry.on_tool_call(cb) fires a ToolCallEvent after every gantry.execute for framework-agnostic logging/metrics; render_result(...) stringifies any tool result. See Observability & tracing.

7. Importing agent_gantry configures no logging. As of 0.8.0 the library attaches a NullHandler and never sets handlers or levels on your behalf, so a default AgentGantry() is silent. Opt into console output with enable_console_logging() (or your own logging config).

Installing

This project uses uv as its package manager; pip works as a fallback. Inside a uv project use uv add; for an ad-hoc environment use uv pip install.

uv add agent-gantry                     # core (in a uv project)
uv add "agent-gantry[nomic]"            # local embeddings (recommended)
uv add "agent-gantry[openai]"           # OpenAI/Azure embeddings + custom OpenAI-compatible base_url
uv add "agent-gantry[agent-frameworks]" # Microsoft AF, LangChain, LangGraph, CrewAI, LlamaIndex, Google ADK
uv add "agent-gantry[lancedb]"          # disk-persistent vector store
uv add "agent-gantry[jev]"              # selection without embeddings (TypeSafe Jev)
uv add "agent-gantry[mcp]"              # MCP client/server
uv add "agent-gantry[a2a]"              # A2A agent
uv add "agent-gantry[all]"              # everything

Not in a uv project? Swap uv add for uv pip install. No uv at all? Fall back to pip install "agent-gantry[...]" — the extras are identical.

The [nomic] extra is the recommended default for getting started — local, free, and accurate enough for production. SimpleEmbedder (the fallback when no embedder extra is installed) is hash-based and only useful for tests.

Minimum-viable code

Every Agent-Gantry program has this shape:

import asyncio
from agent_gantry import AgentGantry

gantry = AgentGantry()

@gantry.register
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"Weather in {city}: sunny"

@gantry.register
def book_flight(origin: str, destination: str) -> str:
    """Book a flight between two cities."""
    return f"Booked {origin} -> {destination}"

async def main():
    await gantry.sync()                                          # embed
    tools = await gantry.retrieve_tools("weather in Paris", limit=3)
    print(tools)  # OpenAI-shape schemas for the top 3 matches

asyncio.run(main())

retrieve_tools(...) returns OpenAI-shape tool schemas by default. Pass dialect="anthropic", "gemini", "agent_framework", etc. to convert.

Microsoft Agent Framework

The native, idiomatic integration. The one-class entry point is AgentFrameworkAdapter; AgentFrameworkAdapter(gantry).context_provider(top_k=...) returns a GantryContextProvider — an AF ContextProvider that runs at before_run (per-run mode) or on every chat-completion round (per-call mode, via a paired middleware). The adapter's other methods build the tool bridge and the approval / observability / tool-choice middleware: .tool_bridge(), .approval_middleware(policy), .observability_middleware(), .tool_choice_middleware(decider). The returned types (GantryContextProvider, GantryToolBridge, the middleware classes) are still importable directly for type annotations.

Per-run mode (default — fixed tool set for one agent.run(...) call)

from agent_framework import Agent
from agent_framework.openai import OpenAIChatClient
from agent_gantry import AgentGantry
from agent_gantry.agent_framework import AgentFrameworkAdapter

gantry = AgentGantry()
# ... register tools, await gantry.sync() ...

provider = AgentFrameworkAdapter(gantry).context_provider(top_k=5)  # -> GantryContextProvider

agent = Agent(
    OpenAIChatClient(),
    "You are a helpful assistant.",
    context_providers=[provider],
)

result = await agent.run("Book me a flight to Tokyo")

Per-call mode (re-runs retrieval every chat round)

Use when the agent reasons in multiple steps and needs different tools at different stages. The chat middleware is required — without it the per-call mode silently degrades to per-run.

from agent_gantry.agent_framework import AgentFrameworkAdapter

provider = AgentFrameworkAdapter(gantry).context_provider(
    top_k=3,
    query_strategy="per_call",
    # Default for per_call is latest_activity: the newest user message or the
    # newest tool result, whichever came last. Pass query_generator=... only
    # for a different rotation (e.g. last_tool_result to always follow the
    # previous tool's output in a pipeline that takes no user turns).
)

agent = Agent(
    OpenAIChatClient(),
    "...",
    context_providers=[provider],
    middleware=[provider.as_chat_middleware()],  # REQUIRED for per_call
)

Or use the one-call helper:

agent = Agent(OpenAIChatClient(), "...")
provider.attach_to(agent)                 # appends provider + middleware in one shot
provider.attach_to(agent, trace=True)     # ...and a console trace of every tool call

trace=True also installs the built-in trace middleware (see Observability & tracing) — the library-owned replacement for hand-rolled @function_middleware logging.

Pinning tools that must always be visible

provider = AgentFrameworkAdapter(gantry).context_provider(
    top_k=5,
    required=["validate_input"],     # MissingRequiredToolError at construction if absent
    always_include=["log_event"],    # warning + skip if absent
    static_tools=[some_af_native_tool],  # tool not registered with gantry
)

required and always_include reference gantry-registered tool names. static_tools is for AF-native @tool callables that live outside the gantry registry — they're appended every round and never filtered.

Static (no per-turn retrieval — bake once)

Use GantryToolBridge directly when the tool set is fixed at construction:

from agent_gantry.integrations.agent_framework_bridge import GantryToolBridge

bridge = GantryToolBridge(gantry)
tools = await bridge.get_tools("customer support tasks", limit=5)
agent = Agent(OpenAIChatClient(), "...", tools=tools)

The bridge also exposes one-call agent constructors: bridge.as_agent(...), bridge.build_handoff_workflow(...), bridge.build_sequential_workflow(...), bridge.build_workflow(...).

Approval middleware + observability

from agent_gantry.core.security import SecurityPolicy
from agent_gantry.integrations.agent_framework_middleware import (
    GantryApprovalMiddleware,
    GantryObservabilityMiddleware,
    GantryToolChoiceMiddleware,
)

policy = SecurityPolicy(require_confirmation=["delete_*", "refund_*"])

# tool_choice modulation: force tool calls for the first N rounds, then auto.
rounds = {"n": 0}
def choose(_ctx):
    rounds["n"] += 1
    return "required" if rounds["n"] <= 5 else "auto"

agent = Agent(
    OpenAIChatClient(),
    "...",
    context_providers=[provider],
    middleware=[
        provider.as_chat_middleware(),
        GantryApprovalMiddleware(policy),
        GantryObservabilityMiddleware(gantry),
        GantryToolChoiceMiddleware(choose),
    ],
)

Native framework adapters

For LangChain, LangGraph, LlamaIndex, CrewAI, Pydantic AI, OpenAI Agents SDK, Haystack, Agno, Strands Agents, DSPy, and Google ADK, use the native <Framework>Adapter class. Its .select(query, limit=...) method selects the top-K relevant tools and returns the framework's native tool objects, with every call still routed through gantry.execute.

from agent_gantry import AgentGantry
from agent_gantry.langchain import LangChainAdapter   # clean per-framework namespace

gantry = AgentGantry()
# ... register tools, await gantry.sync() ...

tools = await LangChainAdapter(gantry).select("email the quarterly report to finance", limit=3)
# hand `tools` straight to a LangChain agent / ChatModel.bind_tools(tools)

Every adapter shares the identical signature <Framework>Adapter(gantry).select(query, *, limit=3, **select_kwargs) and the same per-framework namespace (agent_gantry.<framework>):

Adapters with secondary methods expose them on the same class — there is nothing extra to import.

Need one conversion at a time (you already hold the selected specs)? Use <Adapter>.convert(spec) (a staticmethod) with specs from GantryToolset(gantry).select(query, limit=...).

from agent_gantry.integrations.frameworks import GantryToolset
from agent_gantry.crewai import CrewAIAdapter

specs = await GantryToolset(gantry).select("research and writing", limit=4)
crew_tools = [CrewAIAdapter.convert(s) for s in specs]

Deep per-turn "live" providers

The .select(...) method is static — select once, hand over a fixed list. The live adapter methods hook each framework's own per-turn lifecycle so Gantry re-selects tools on every turn, matching GantryContextProvider depth for Microsoft Agent Framework. Construct the adapter from the per-framework namespace, then call the live method.

from agent_gantry.llamaindex import LlamaIndexAdapter
agent = LlamaIndexAdapter(gantry).function_agent(llm)   # re-selects tools each step

The returned live objects keep their classes, importable from the framework's *_live module for isinstance checks: GantryToolRetriever (llamaindex_live), GantryToolset (pydantic_ai_live), GantryAgentSession (openai_agents_live) and GantryStrandsToolHook (strands_live). Google ADK and LangGraph return plain framework objects (a before_model_callback callable and a compiled agent), so there is no Gantry class to check for.

Frameworks whose tool list is fixed at agent construction (CrewAI, Agno, Haystack, DSPy) can't re-advertise tools mid-run. Build a self-rebuilding agent with <Adapter>(gantry).agent_builder(...) (Haystack: HaystackAdapter(gantry).tool_invoker_builder(...); DSPy: DSPyAdapter(gantry).agent_builder(signature, ...), since dspy.ReAct needs a task signature); it re-selects and rebuilds on each top-level call. For a one-shot fresh slice of native tools, call <Adapter>(gantry).live_tools(query) (async, not available on DSPy — use .select(query) instead).

Selection without embeddings (Jev)

Added in 0.16.0. JevSelector replaces embed-then-search entirely: it puts the catalogue to TypeSafe's Jev decision model and asks, per entry, whether it is relevant. No embedder, no vector store, no sync(). JevReranker instead refines an ordinary semantic shortlist. Both are optional, both are off by default, and both live behind the jev extra.

uv add "agent-gantry[jev]"
export TYPESAFE_API_KEY=...

from agent_gantry import AgentGantry, JevSelector

gantry = AgentGantry(selector=JevSelector(threshold=0.4))
# register tools as usual; no sync() needed on the selector path
tools = await gantry.retrieve_tools("refund the customer's order", limit=3)

It covers all three catalogues, not just tools: retrieve_tools, retrieve_skills and retrieve_mcp_servers all consult the selector first.

As a reranker instead, keeping semantic retrieval:

from agent_gantry import AgentGantry, JevReranker

gantry = AgentGantry(reranker=JevReranker())

When each one fits. A selector sends every candidate on every query, so cost grows with catalogue size rather than staying flat. That suits tens to a few hundred entries. Above that, keep semantic retrieval and put the reranker over its shortlist — only the shortlist is sent, so it works at any catalogue size. SelectorConfig.group_after handles the middle ground by narrowing to the best namespaces before scoring their members.

What it buys. Measured against a properly configured embedder, recall is a tie. The difference is precision: 1–3 spurious tools against 16 on the same set. That is the context-window saving, not "finds better tools". A selector can also answer none of these, which top-k cannot — with limit=3 a vector search always returns three tools whether or not any of them fit.

The sharper win is distinctions an embedder cannot make, because it compares text rather than reasoning about fit. Measured live against jev-1.13.0, with a get_weather tool described as "get the current weather":

"current weather" and "weather forecast" embed almost on top of each other, so a vector search hands the agent a tool that cannot answer the question. Jev declined instead. If a user is choosing between the two approaches, that — not the token count — is the argument.

Both fail open, and neither raises into a caller's retrieval path — but they fail open differently, and the difference matters if you are choosing between them.

A provider that is unavailable, rate-limited or slow costs precision and never the catalogue: the selector reports a fallback and retrieval takes the semantic path, while the reranker returns the vector-search order untouched, truncated to top_k.

A provider that answers for only some of the candidates is not the same case. The selector treats it as a failed pass and falls back — a candidate that was never scored is indistinguishable from one scored zero, so ranking the answered subset would be a confident guess. The reranker keeps the pass: answered candidates are ordered by probability, unanswered ones keep their search rank below all of them, and the result is truncated to top_k. That last step is the caveat — where more candidates were sent than top_k, a high-ranked candidate the model simply did not answer for can be pushed out of the result. It is a precision loss rather than a correctness one, and it cannot happen on the selector path.

If you see selection quietly stop working, check the logs for Jev selection failed, falling back: — that is the library declining to guess, not a silent failure.

Query constraints are applied before anything is sent — deprecation, namespaces, capabilities, sources and circuit-breaker health all go through the router's own filter (SemanticRouter.filter_tools()), so a selector can never surface a tool the semantic path would hide.

Debugging. SelectionResult carries scores for every candidate, not just the winners, so you can see exactly what a threshold would keep and cut before committing to it — see examples/routing/jev_threshold_tuning_demo.py, which sweeps thresholds over a labelled query set.

Multi-turn re-selection (ToolRefresher)

ToolRefresher generalizes the per-call retrieval pattern to any framework without a native live hook: re-rank the whole registry on every turn so the agent can pivot tools as the task changes.

from agent_gantry.integrations import ToolRefresher

refresher = ToolRefresher(gantry, limit=3, dialect="openai")

# Each turn, pass the running message list. Selection is recomputed fresh and
# tools already used are deprioritized so the agent keeps moving forward.
tools = await refresher.refresh(messages)        # dialect schemas
specs = await refresher.refresh_specs(messages)  # ToolSpec objects

Used-tool tracking reads the name from wherever the SDK keeps it — an OpenAI tool_calls entry, an Anthropic tool_use block, a LangChain ToolMessage or a {"role": "tool", "name": ...} dict (agent_gantry.query.tool_names_used).

The default query generator (agent_gantry.query.latest_activity) is recency-aware, so one refresher serves both styles with no config:

• Autonomous agents / tool pipelines — newest message is a tool result, so that result's content selects the next tool (fetch → clean → train → evaluate → report).
• Conversational agents — newest message is the user's, so their new request drives selection (weather → flights → hotel → email).

Force one behaviour with query_generator=last_user_text or last_tool_result.

Schema-only adapter (fetch_framework_tools)

The legacy adapter returns OpenAI-shape JSON schemas (not native objects) and supports a smaller, hyphen/underscore-sensitive name set: "langgraph", "crew_ai", "google_adk", "strands", "agent_framework". Prefer the <Framework>Adapter classes for the frameworks listed above; reach for this only when you want raw schemas (e.g. a framework that just wants OpenAI tool dicts).

from agent_gantry.integrations import fetch_framework_tools

schemas = await fetch_framework_tools(
    gantry, "code execution", framework="google_adk", limit=3
)

LLM-SDK direct

For a minimal stack — no agent framework — there are two paths.

Per-SDK adapter classes (explicit, no default-gantry binding). One class per provider, each baking in the right schema dialect:

from openai import AsyncOpenAI
from agent_gantry import AgentGantry
from agent_gantry.openai import OpenAIAdapter

gantry = AgentGantry()
# ... register tools, await gantry.sync() ...

client = AsyncOpenAI()
tools = await OpenAIAdapter(gantry).tools("weather in Paris", limit=3)  # OpenAI chat-completions schemas
resp = await client.chat.completions.create(
    model="gpt-5.5",
    messages=[{"role": "user", "content": "weather in Paris?"}],
    tools=tools,
)

<Provider>Adapter(gantry).tools(query, limit=n) is equivalent to await gantry.retrieve_tools(query, limit=n, dialect="<provider>") — the adapter just bakes the dialect in.

Decorator (@with_semantic_tools) when you'd rather bind a default gantry and have tools injected automatically:

from openai import AsyncOpenAI
from agent_gantry import AgentGantry, set_default_gantry, with_semantic_tools

gantry = AgentGantry()
set_default_gantry(gantry)

@gantry.register
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return "..."

client = AsyncOpenAI()

@with_semantic_tools(limit=3)   # default dialect="openai"
async def chat(prompt: str, *, tools=None):
    return await client.chat.completions.create(
        model="gpt-5.5",
        messages=[{"role": "user", "content": prompt}],
        tools=tools,
    )

await chat("weather in Paris?")

Dialect support: dialect="anthropic", "gemini", "mistral", "groq", "openai_responses". Each emits the right schema for that provider; the call shape stays identical.

Mistral note: the official mistralai SDK is quarantined on PyPI. Use AsyncOpenAI pointed at https://api.mistral.ai/v1 — Mistral's API is OpenAI-compatible. With agent-gantry, the OpenAI dialect (or MistralAdapter) works directly.

OpenAI-compatible custom endpoints (Requesty, OpenRouter, Together, vLLM, …) are first-class: pass api_base in the EmbedderConfig or set OPENAI_BASE_URL, and OpenAIEmbedder forwards it to the client.

MCP

Expose Gantry as an MCP server (Claude Desktop, Claude Code, Cline, any remote MCP client):

gantry = AgentGantry()
# ... register tools, await gantry.sync() ...
await gantry.serve_mcp()                                   # stdio, dynamic meta-tools
await gantry.serve_mcp("http", port=8000, path="/mcp")     # Streamable HTTP (remote clients)
await gantry.serve_mcp("sse", port=8000)                   # legacy SSE transport
await gantry.serve_mcp(mode="hybrid", expose=["get_weather", "billing.search"])

• mode="dynamic" (default) exposes two meta-tools, find_relevant_tools and execute_tool — the client semantic-searches Gantry on demand, so its tool list stays two entries long however many tools you register (~90% smaller than mode="static"). find_relevant_tools reports each match's exact tool_name (qualified as namespace.name outside the default namespace) and JSON Schema parameters.
• mode="hybrid" lists the tools named in expose directly and keeps the meta-tools for everything else — for the handful of tools a client should always see.
• mode="static" lists every tool.
• From the shell: agent-gantry serve-mcp --module my_app.tools [--transport http|sse] [--mode hybrid --expose get_weather]. For Claude Desktop, put {"command": "agent-gantry", "args": ["serve-mcp", "--module", "my_app.tools"]} under mcpServers.
• To mount into an existing ASGI service: create_mcp_server(gantry).streamable_http_app("/mcp") returns a Starlette app (keep its lifespan). Pass allowed_hosts=[...] / allowed_origins=[...] to serve_mcp to enable DNS-rebinding protection when binding beyond loopback.

Consume external MCP servers — local (stdio) or remote (Streamable HTTP / SSE):

from agent_gantry.schema.config import MCPServerConfig

await gantry.add_mcp_server(MCPServerConfig(
    name="filesystem",
    command=["npx", "-y", "@modelcontextprotocol/server-filesystem"],
    args=["--path", "/tmp"],
    namespace="fs",
))
await gantry.add_mcp_server(MCPServerConfig(
    name="search",
    url="https://mcp.example.com/mcp",                 # Streamable HTTP (inferred from url)
    headers={"Authorization": "Bearer ..."},           # auth goes in headers; never logged
    namespace="web",
))
await gantry.add_mcp_server(MCPServerConfig(name="legacy", url="https://old.example.com/sse", transport="sse"))

Exactly one of command / url is required. Discovered tool names are normalised to Gantry's snake_case (searchWeb → search_web, get-weather → get_weather); the server is still called by its original name (kept in metadata["mcp_tool_name"]), and short/long descriptions are padded/truncated rather than rejected, so one unconventional tool never fails a server's whole discovery. Same url=/headers=/transport= keywords on gantry.register_mcp_server(...) for dynamic server selection (see cookbook Recipe 7).

A2A

gantry.serve_a2a(host="0.0.0.0", port=8080)
# Agent Card at: http://localhost:8080/.well-known/agent.json

Skills exposed: tool_discovery (semantic search) and tool_execution (run a tool).

CLI

The agent-gantry command ships with the package. Inside a uv project, prefix with uv run (e.g. uv run agent-gantry list); the bare agent-gantry form works when the package is on PATH (pip install or an activated venv).

Every inspection command works on your registry: --module pkg.tools (or pkg.tools:attr) names the module holding your AgentGantry instance and the CLI uses that instance directly — its embedder, its vector store. Without --module a three-tool demo registry is used (a note goes to stderr).

uv run agent-gantry list --module my_app.tools                     # list registered tools
uv run agent-gantry search "refund order" --module my_app.tools     # semantic search, --limit 3
uv run agent-gantry lint --module my_app.tools                      # detect tool-description authoring mistakes
uv run agent-gantry lint --source src/                              # scan files: inverted score_threshold, unclosed gantries
uv run agent-gantry sim toolA toolB --module my_app.tools           # cosine similarity between two tools
uv run agent-gantry sync --dry-run --module my_app.tools            # which tools would (re-)embed and why
uv run agent-gantry sync --prune --module my_app.tools              # also drop stored tools no longer registered
uv run agent-gantry serve-mcp --module my_app.tools                 # MCP server over stdio (Claude Desktop / Claude Code)
uv run agent-gantry serve-mcp --module my_app.tools --transport http --mode hybrid --expose get_weather
uv run agent-gantry install-skill --claude                          # install THIS skill into ~/.claude/skills

lint flags three patterns that silently degrade routing quality:

1. Descriptions that mention other registered tools (embedding pulls them toward each other).
2. Pairs of tools with >0.85 cosine similarity (probably should be one tool, or differentiated).
3. Tags that appear on more than half the registry (low discriminative value).

Observability & tracing

As of 0.8.0 the library ships the tracing/logging glue you'd otherwise hand-roll. Reach for these before writing custom middleware.

Console trace — one readable line per tool call

provider.trace() returns an AF function middleware; attach_to(agent, trace=True) installs it (plus the per-call retrieval middleware) in one call. For every tool the model invokes it prints the round, the call, the router's surfaced set, and a preview of the result:

provider = AgentFrameworkAdapter(gantry).context_provider(top_k=3, query_strategy="per_call")
agent = Agent(OpenAIChatClient(), "...")
provider.attach_to(agent, trace=True)
# >>> round 1: generate_password({'length': 20})  [surfaced: generate_password:0.62, ...]
# <<< round 1: generate_password -> Xk9$mP2v...

provider.trace(render=False, printer=logger.info, limit=200) tunes it: skip the result line, redirect the sink (e.g. to a logger), or change the preview length.

Per-round selection history

provider.last_selection is a single mutable slot. provider.selections is the per-round history for the current run (oldest first, bounded; reset at the start of each agent.run), so you can audit what was surfaced at each step of a run, not just the last:

for i, decision in enumerate(provider.selections, start=1):
    print(f"round {i}: {decision.summary()}")

Framework-agnostic tool-call events

gantry.on_tool_call(callback) fires after every gantry.execute (and once per call in execute_batch) with a ToolCallEvent. Because execute is the single choke point every framework adapter routes through, this gives logging/metrics across LangChain, CrewAI, AF, direct calls — with no per-framework middleware:

from agent_gantry import ToolCallEvent

def log_call(event: ToolCallEvent) -> None:
    status = "ok" if event.ok else f"FAILED ({event.result.error})"
    print(f"{event.tool_name} -> {status} ({event.latency_ms:.0f} ms)")

unsubscribe = gantry.on_tool_call(log_call)   # sync or async callbacks; returns an unsubscribe fn

Callbacks are error-isolated — a raising listener never breaks the tool run — and failed tools still emit an event (event.ok is False). ToolCallEvent carries .call, .result, and the convenience accessors .tool_name, .status, .ok, .latency_ms.

Batch timing: events fire immediately for each gantry.execute, but for gantry.execute_batch they're emitted after the whole batch finishes (one per call, paired by index). latency_ms stays accurate; only delivery is batched — relevant if you timestamp events for a latency dashboard.

Rendering tool results

render_result(result, *, limit=None, collapse_whitespace=False) turns any result — including AF Content-block lists, bytes, dicts, or arbitrary objects — into readable text for logs/UIs. The trace middleware uses it internally.

Logging

The library configures no logging (a NullHandler is attached at import). To see Gantry's own INFO lines on the console, opt in once:

from agent_gantry import enable_console_logging
enable_console_logging()   # attaches a console handler + sets the agent_gantry level

Telemetry (ConsoleTelemetryAdapter, the default) still emits structured records; they flow to whatever handlers your app configured. There's nothing to silence anymore — a default AgentGantry() is quiet.

Skills (procedural memory, retrieved by meaning)

Skills are how-tos, patterns and procedures that are injected into the prompt, never executed. Gantry embeds each skill's metadata once and retrieves only the ones relevant to the current prompt — the same top-k routing it applies to tools, so a library of hundreds of skills costs a few hundred tokens per turn instead of every skill's description.

from agent_gantry import AgentGantry, Skill, SkillCategory

gantry = AgentGantry()
await gantry.add_skill(Skill(
    name="api_pagination",
    description="How to implement cursor-based pagination for API endpoints",
    content="Use cursor-based pagination: return an opaque cursor per page ...",
    category=SkillCategory.HOW_TO, tags=["api", "pagination"], related_tools=["fetch_page"],
))

# Any Agent Skills directory (Claude Code / Claude Agent SDK `SKILL.md` format) loads as-is:
await gantry.add_skills_from_directory("~/.claude/skills")          # name/description/body per SKILL.md

system_prompt += await gantry.retrieve_skills_as_prompt(user_prompt, limit=3)   # "" when nothing matches
results = await gantry.retrieve_skills("paginate the users endpoint", limit=3, namespace="docs", category="how_to")

retrieve_skills_as_prompt formats each hit with Skill.to_prompt_text() (a heading, the category, the full content, related tools). Skill content is injected verbatim — register skills only from sources you trust. Works with the default in-memory store and LanceDB; a model switch re-embeds stored skills automatically.

Debugging routing

When a user says "the LLM doesn't see my tool" / "my surface is empty" / "wrong tools are being selected", use these in order:

1. provider.last_selection — what just happened

provider = GantryContextProvider(gantry, top_k=5)
result = await agent.run("...")

decision = provider.last_selection
print(decision.summary())           # one-line: query + top scores
print(decision.injected)            # tool names that made it to the LLM
for c in decision.candidates:
    print(c.name, c.score, c.kept)  # full ranked list, kept/dropped flag

RetrievalDecision carries: the query, every candidate the gantry returned, the threshold mode used, the effective numeric cutoff, and the final injected list. For the per-round history (not just the latest), use provider.selections — see Observability & tracing.

2. provider.dry_run_retrieve(query) — same code path as live, no agent

decision = await provider.dry_run_retrieve("find boundaries in OCR text")
for c in decision.candidates:
    print(f"{c.score:.3f}  {c.qualified_name}  kept={c.kept}")

This uses the exact same threshold, top_k, and query_kwargs as the live middleware — so the answer is "what the LLM would see for that query".

3. verbose=True — one-line INFO log per round

provider = GantryContextProvider(gantry, top_k=5, verbose=True)
# logs: gantry: query="..." → top5: [tool_a:0.61, tool_b:0.58, ...]

4. score_threshold filtered everything?

When the threshold drops all candidates, the bridge logs a WARNING with the top scores so you can tell "threshold issue" from "relevance issue". Default is 0.0. If a user has set score_threshold=0.3 (the legacy default) on a long pipeline query, lower it or switch to relative:

provider = GantryContextProvider(
    gantry,
    score_threshold="relative:0.8",   # keep anything within 80% of top score
)

5. Long queries silently degrade routing

Long imperative scaffolding ("Please run this five-step pipeline. Use a different tool for each step…") dilutes the embedding. Strip it with keyword_focused:

from agent_gantry.query import keyword_focused, truncated, last_user_text

provider = GantryContextProvider(
    gantry,
    query_strategy="per_call",
    query_generator=keyword_focused,       # drops scaffolding tokens
)

# Or cap the query length, biasing toward the latest tool output:
provider = GantryContextProvider(
    gantry,
    query_strategy="per_call",
    query_generator=truncated(last_user_text, max_chars=200, keep="tail"),
)

6. Lint the registry — author-side bugs

agent-gantry lint

Or programmatically:

analysis = await gantry.analyze_registry()
print(analysis.format_text())

This catches the headline mistakes: a tool description that names another tool (which pulls it toward the wrong queries), pairs of tools that are too similar to disambiguate, and tags that are too generic.

Common pitfalls

Persistent embedding cache

Re-embedding costs API spend at every cold start. Wrap any embedder:

from agent_gantry import AgentGantry
from agent_gantry.adapters.embedders.openai import OpenAIEmbedder
from agent_gantry.adapters.embedders.cached import CachedEmbedder
from agent_gantry.schema.config import EmbedderConfig

base = OpenAIEmbedder(EmbedderConfig(type="openai", model="text-embedding-3-large"))
embedder = CachedEmbedder(base)   # default: ~/.cache/agent_gantry/embeddings.sqlite

gantry = AgentGantry(embedder=embedder)

Cache is keyed by (embedder_id, sha256(text)) — different models / dimensions never collide.

Query generators reference

Both query_strategy="per_call" and ToolRefresher default to latest_activity, which serves autonomous and conversational agents alike: a new user message pivots the surface, and while the agent chains tools the previous result drives the next selection. fallback_chain(last_tool_result, last_user_text) — the former per_call default — prefers a tool result anywhere in the history, so a session that carries an earlier run's result stops following new user requests; use it only for a fixed pipeline with no user turns in between.

Every generator reads the common SDK shapes: role/content dicts, OpenAI Chat Completions tool messages, OpenAI Responses items (function_call / function_call_output), Anthropic tool_result turns (a user-role message of result blocks), LangChain messages and Agent Framework contents.

API reference quick-card

from agent_gantry import (
    AgentGantry,                  # main facade
    GantryContextProvider,        # AF native provider (per-run / per-call)
    MissingRequiredToolError,     # raised when required=[...] tool absent
    RetrievalDecision,            # introspection: what the bridge surfaced
    RetrievalCandidate,           # one row in RetrievalDecision.candidates
    with_semantic_tools,          # decorator for plain LLM SDK calls
    set_default_gantry,           # bind a gantry to with_semantic_tools
    create_default_gantry,        # quick factory (auto-picks Nomic if available)
    render_result,                # stringify any tool result (incl. AF Content lists)
    enable_console_logging,       # opt into console output (lib configures no logging)
    ToolCall, ToolResult,
    ToolCallEvent,                # delivered to gantry.on_tool_call(callback)
    ToolQuery, ConversationContext,
    ToolCapability, ToolCost, ToolDefinition, ToolHealth, ToolSource,
    Skill, SkillCategory, SkillSearchResult,   # procedural memory (see "Skills")
)
# Skills: gantry.add_skill / add_skills / add_skills_from_directory(path)
#         gantry.retrieve_skills(query) / retrieve_skills_as_prompt(query)
# Selection (0.16.0+, needs the `jev` extra):
#   AgentGantry(selector=JevSelector(threshold=0.4))  -> no embedder/vector store/sync
#   AgentGantry(reranker=JevReranker())               -> refine a semantic shortlist
#   SelectionResult.scores                            -> every candidate, not just winners
# MCP:    gantry.serve_mcp(transport="stdio"|"http"|"sse", mode=..., expose=[...])
#         reset_sse_shutdown_latch()  -> release sse_starlette's process-global
#           shutdown latch between servers in one process; without it a second
#           EventSourceResponse is cancelled between headers and body (hangs).
#           Needs the `mcp` extra; raises AttributeError without it.
#         gantry.add_mcp_server(MCPServerConfig(command=[...] | url="https://...", headers=...))
#         gantry.register_mcp_server(name, command=None, *, url=None, description=...)
#         gantry.sync(prune=True) drops stored tools that are no longer registered
# Observability built-ins (0.8.0+):
#   gantry.on_tool_call(cb)            -> framework-agnostic ToolCallEvent stream
#   provider.trace() / attach_to(..., trace=True)  -> console trace middleware
#   provider.selections                -> per-round RetrievalDecision history
from agent_gantry.integrations import (
    GantryToolBridge,             # static AF bridge + workflow builders
    GantryApprovalMiddleware,
    GantryObservabilityMiddleware,
    GantryToolChoiceMiddleware,   # round-by-round tool_choice modulation
    ToolRefresher,                # framework-agnostic multi-turn re-selection
    fetch_framework_tools,        # legacy schema-only adapter (OpenAI-shape)
)
from agent_gantry.integrations.frameworks import (
    GantryToolset,                # selection core behind every <Framework>Adapter
    ToolSpec,                     # framework-neutral tool handle (.ainvoke/.invoke)
    spec_from_tool,
)
# One <Framework>Adapter class per framework, from the clean per-framework
# namespace. Each exposes .select(query, limit=...) (async) and
# .convert(spec) (staticmethod):
from agent_gantry.langchain import LangChainAdapter
from agent_gantry.langgraph import LangGraphAdapter
from agent_gantry.llamaindex import LlamaIndexAdapter
from agent_gantry.crewai import CrewAIAdapter
from agent_gantry.pydantic_ai import PydanticAIAdapter
from agent_gantry.openai_agents import OpenAIAgentsAdapter
from agent_gantry.haystack import HaystackAdapter
from agent_gantry.agno import AgnoAdapter
from agent_gantry.strands import StrandsAdapter
from agent_gantry.dspy import DSPyAdapter
from agent_gantry.google_adk import GoogleADKAdapter
from agent_gantry.agent_framework import AgentFrameworkAdapter   # Microsoft Agent Framework
# Per-SDK LLM adapters (dialect baked in): .tools(query, limit=...)
from agent_gantry.openai import OpenAIAdapter
from agent_gantry.anthropic import AnthropicAdapter
from agent_gantry.gemini import GeminiAdapter
from agent_gantry.groq import GroqAdapter
from agent_gantry.vertexai import VertexAIAdapter
from agent_gantry.mistral import MistralAdapter
from agent_gantry.query import (
    last_user_text, last_assistant_text, last_tool_result,
    concatenate_recent, fallback_chain,
    keyword_focused, truncated, latest_activity,
)
from agent_gantry.adapters.embedders.cached import CachedEmbedder
from agent_gantry.skills import load_skills_from_directory, load_skill   # Agent Skills (SKILL.md) loaders
from agent_gantry.servers.mcp_server import create_mcp_server          # .streamable_http_app() / .sse_app()
from agent_gantry.utils.registry_linter import (
    analyze_registry, pairwise_similarity, RegistryAnalysis,
)

For detailed reference on individual modules, see references/ next to this file.