Skip to main content

​
Overview

These advanced patterns demonstrate production-ready implementations for real-world applications.

​
Tool Runner Hooks

Intercept and modify agent behavior at different stages:
examples/tool-runner-hooks/tool_runner_hooks.py
import asyncio

from fast_agent import FastAgent
from fast_agent.agents.agent_types import AgentConfig
from fast_agent.agents.tool_agent import ToolAgent
from fast_agent.agents.tool_runner import ToolRunnerHooks
from fast_agent.context import Context
from fast_agent.interfaces import ToolRunnerHookCapable
from fast_agent.types import PromptMessageExtended

def get_video_call_transcript(video_id: str) -> str:
    return "Assistant: Hi, how can I assist you today?\n\nCustomer: Hi, I wanted to ask you about last invoice I received..."

class HookedToolAgent(ToolAgent, ToolRunnerHookCapable):
    def __init__(
        self,
        config: AgentConfig,
        context: Context | None = None,
    ):
        tools = [get_video_call_transcript]
        super().__init__(config, tools, context)
        self._hooks = ToolRunnerHooks(
            before_llm_call=self._add_style_hint,
            after_tool_call=self._log_tool_result,
        )
    
    @property
    def tool_runner_hooks(self) -> ToolRunnerHooks | None:
        return self._hooks
    
    async def _add_style_hint(self, runner, messages: list[PromptMessageExtended]) -> None:
        if runner.iteration == 0:
            runner.append_messages("Keep the answer to one short sentence.")
    
    async def _log_tool_result(self, runner, message: PromptMessageExtended) -> None:
        if message.tool_results:
            tool_names = ", ".join(message.tool_results.keys())
            print(f"[hook] tool results received: {tool_names}")

fast = FastAgent("Example Tool Use Application (Hooks)")

@fast.custom(HookedToolAgent)
async def main() -> None:
    async with fast.run() as agent:
        await agent.default.generate(
            "What is the topic of the video call no.1234?",
        )
        await agent.interactive()

if __name__ == "__main__":
    asyncio.run(main())
Available hooks: before_llm_call, after_llm_call, before_tool_call, after_tool_call, on_error

​
Hook Use Cases

Track tool usage and performance:
async def log_tool_usage(runner, message: PromptMessageExtended) -> None:
    if message.tool_results:
        for tool_name, result in message.tool_results.items():
            logger.info(f"Tool: {tool_name}, Duration: {result.duration}ms")
            metrics.increment(f"tool.{tool_name}.calls")
Sanitize or format responses:
async def sanitize_response(runner, messages: list[PromptMessageExtended]) -> None:
    for msg in messages:
        if msg.role == "assistant" and msg.content:
            # Remove sensitive information
            msg.content = sanitize(msg.content)
Monitor token usage:
async def track_tokens(runner, message: PromptMessageExtended) -> None:
    if hasattr(message, 'usage'):
        cost = calculate_cost(message.usage)
        tracker.add_cost(cost)
        logger.info(f"Request cost: ${cost:.4f}")
Implement custom retry behavior:
async def retry_on_error(runner, error: Exception) -> None:
    if isinstance(error, RateLimitError):
        await asyncio.sleep(1)
        # Retry will happen automatically
    else:
        raise error

​
RAG Integration

Integrate Retrieval-Augmented Generation with Google Vertex AI:
examples/rag/vertex-rag.py
import asyncio
import google.auth
import vertexai
from vertexai import rag
from fast_agent import FastAgent
from fast_agent.config import get_settings

CONFIG_PATH = "fastagent.secrets.yaml"
_settings = get_settings(CONFIG_PATH)
_vertex_ai = getattr(_settings.google, "vertex_ai", {}) if _settings.google else {}
PROJECT_ID = _vertex_ai.get("project_id")
LOCATION = _vertex_ai.get("location")
EMBEDDING_MODEL = "text-embedding-005"

def mini_rag(query: str, drive_id: str, top_k: int) -> object:
    """Query RAG corpus for relevant documents."""
    vertexai.init(project=PROJECT_ID, location=LOCATION)
    
    # Create or get existing corpus
    paths = [f"https://drive.google.com/drive/folders/{drive_id}"]
    existing_corpus = None
    for corpus in rag.list_corpora():
        if corpus.display_name and drive_id in corpus.display_name:
            existing_corpus = corpus
            break
    
    if not existing_corpus:
        # Create new corpus and import files
        embedding_model_config = rag.RagEmbeddingModelConfig(
            vertex_prediction_endpoint=rag.VertexPredictionEndpoint(
                publisher_model=f"publishers/google/models/{EMBEDDING_MODEL}"
            )
        )
        rag_corpus = rag.create_corpus(
            display_name=f"Corpus {drive_id}",
            backend_config=rag.RagVectorDbConfig(
                rag_embedding_model_config=embedding_model_config
            ),
        )
        rag.import_files(
            rag_corpus.name,
            paths,
            transformation_config=rag.TransformationConfig(
                chunking_config=rag.ChunkingConfig(
                    chunk_size=512,
                    chunk_overlap=100,
                ),
            ),
        )
    else:
        rag_corpus = existing_corpus
    
    # Query the corpus
    rag_retrieval_config = rag.RagRetrievalConfig(
        top_k=top_k,
        filter=rag.Filter(vector_distance_threshold=0.5),
    )
    return rag.retrieval_query(
        rag_resources=[rag.RagResource(rag_corpus=rag_corpus.name)],
        text=query,
        rag_retrieval_config=rag_retrieval_config,
    )

fast = FastAgent("Google Vertex RAG")

@fast.agent(
    name="vertex_rag",
    function_tools=[mini_rag],
)
async def main():
    async with fast.run() as agent:
        result = await agent(
            "Produce a short top 5 prioritized list about customer pain points. From RAG, select 50 relevant chunks."
        )
        print(result)

if __name__ == "__main__":
    asyncio.run(main())
Configure Vertex AI credentials in fastagent.secrets.yaml:
google:
  vertex_ai:
    enabled: true
    project_id: your-project-id
    location: us-central1

​
FastAPI Integration

​
Simple Integration

Basic FastAPI integration with agent lifecycle management:
examples/fastapi/fastapi-simple.py
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from fast_agent import FastAgent

# Create FastAgent without parsing CLI args
fast = FastAgent("fast-agent demo", parse_cli_args=False, quiet=True)

@fast.agent(name="helper", instruction="You are a helpful AI Agent.", default=True)
async def decorator():
    pass

# Keep FastAgent running for the app lifetime
@asynccontextmanager
async def lifespan(app: FastAPI):
    async with fast.run() as agents:
        app.state.agents = agents
        yield

app = FastAPI(lifespan=lifespan)

class AskRequest(BaseModel):
    message: str

class AskResponse(BaseModel):
    response: str

@app.post("/ask", response_model=AskResponse)
async def ask(req: AskRequest) -> AskResponse:
    try:
        result = await app.state.agents.send(req.message)
        return AskResponse(response=result)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))
Run with: 
uvicorn fastapi-simple:app --reload

​
Advanced Integration

Manual lifecycle control with Core API:
examples/fastapi/fastapi-advanced.py
from contextlib import asynccontextmanager
from fastapi import Body, FastAPI, HTTPException
from fast_agent import PromptMessageExtended
from fast_agent.agents import McpAgent
from fast_agent.agents.agent_types import AgentConfig
from fast_agent.core import Core
from fast_agent.core.direct_factory import get_model_factory

core = Core(name="fast-agent core")

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Manual lifecycle control
    await core.initialize()
    
    cfg = AgentConfig(
        name="core_agent",
        instruction="You are a helpful AI Agent.",
    )
    
    agent = McpAgent(config=cfg, context=core.context)
    await agent.initialize()
    
    llm_factory = get_model_factory(core.context, model=cfg.model)
    await agent.attach_llm(llm_factory)
    
    app.state.agent = agent
    try:
        yield
    finally:
        try:
            await agent.shutdown()
        finally:
            await core.cleanup()

app = FastAPI(lifespan=lifespan)

@app.post("/ask", response_model=PromptMessageExtended)
async def ask(body: str = Body(..., media_type="text/plain")) -> PromptMessageExtended:
    try:
        # Call generate() to return the full multipart message
        result = await app.state.agent.generate(body)
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))
The advanced pattern gives you full control over agent initialization and cleanup, useful for custom configurations.

​
OpenTelemetry Integration

Monitor your agents with OpenTelemetry:
fastagent.config.yaml
observability:
  otel:
    enabled: true
    endpoint: "http://localhost:4318"
    service_name: "fast-agent"
    traces:
      enabled: true
      sample_rate: 1.0
    metrics:
      enabled: true
Instrument your code: 
from fast_agent.observability import get_tracer, get_meter

tracer = get_tracer(__name__)
meter = get_meter(__name__)

# Create metrics
request_counter = meter.create_counter(
    "agent.requests",
    description="Number of agent requests"
)

@fast.agent(name="monitored_agent")
async def main():
    async with fast.run() as agent:
        with tracer.start_as_current_span("agent.process"):
            request_counter.add(1)
            result = await agent.send("Hello")
            return result

​
Agent-to-Agent Communication

Connect agents across different processes or machines:
examples/a2a/server.py
import asyncio
from fast_agent import FastAgent

fast = FastAgent("A2A Server")

@fast.agent(
    name="worker",
    instruction="You are a worker agent that processes tasks.",
    servers=["filesystem"],
)
async def main():
    # Start A2A server
    await fast.start_a2a_server(host="0.0.0.0", port=8765)

if __name__ == "__main__":
    asyncio.run(main())
Connect from another agent:
examples/a2a/agent_executor.py
import asyncio
from fast_agent import FastAgent

fast = FastAgent("A2A Client")

@fast.agent(
    name="coordinator",
    instruction="You coordinate tasks with remote workers.",
    a2a_connections=[
        {"name": "worker", "url": "ws://localhost:8765"}
    ],
)
async def main():
    async with fast.run() as agent:
        # Call remote agent as a tool
        result = await agent.coordinator.send(
            "Process this file using the worker agent"
        )
        print(result)

if __name__ == "__main__":
    asyncio.run(main())
A2A communication is experimental and subject to change. Use in production with caution.

​
Custom Model Providers

Integrate custom LLM providers: 
from fast_agent.llm.base import BaseLLM, LLMConfig
from fast_agent.types import PromptMessageExtended

class CustomLLM(BaseLLM):
    def __init__(self, config: LLMConfig):
        super().__init__(config)
        # Initialize your custom provider
        self.client = CustomClient(api_key=config.api_key)
    
    async def generate(
        self,
        messages: list[PromptMessageExtended],
        **kwargs
    ) -> PromptMessageExtended:
        # Implement your generation logic
        response = await self.client.complete(messages)
        return self._format_response(response)

# Register the provider
from fast_agent.llm.model_factory import ModelFactory

ModelFactory.register_provider("custom", CustomLLM)

# Use in your agent
@fast.agent(model="custom:my-model")
async def main():
    async with fast.run() as agent:
        await agent.interactive()

​
Production Best Practices

Implement comprehensive error handling:
from fast_agent.exceptions import AgentError, ToolError, LLMError

try:
    result = await agent.send(message)
except ToolError as e:
    logger.error(f"Tool failed: {e.tool_name} - {e.message}")
    # Handle tool failure
except LLMError as e:
    logger.error(f"LLM error: {e}")
    # Handle LLM failure
except AgentError as e:
    logger.error(f"Agent error: {e}")
    # Handle general agent error
Implement rate limiting for API calls:
from aiolimiter import AsyncLimiter

rate_limiter = AsyncLimiter(max_rate=10, time_period=1)  # 10 requests per second

async def rate_limited_send(agent, message):
    async with rate_limiter:
        return await agent.send(message)
Use environment variables and secret managers:
fastagent.secrets.yaml
llm:
  anthropic:
    api_key: "${ANTHROPIC_API_KEY}"
  openai:
    api_key: "${OPENAI_API_KEY}"

mcp:
  servers:
    brave-search:
      env:
        BRAVE_API_KEY: "${BRAVE_API_KEY}"
Load from AWS Secrets Manager:
import boto3
import json

def load_secrets():
    client = boto3.client('secretsmanager')
    response = client.get_secret_value(SecretId='fast-agent/prod')
    return json.loads(response['SecretString'])
Implement health check endpoints:
@app.get("/health")
async def health_check():
    try:
        # Check agent is responsive
        await app.state.agent.send("ping")
        return {"status": "healthy", "agent": "ready"}
    except Exception as e:
        raise HTTPException(status_code=503, detail=str(e))

@app.get("/ready")
async def readiness_check():
    if not app.state.agent:
        raise HTTPException(status_code=503, detail="Agent not initialized")
    return {"status": "ready"}
Handle shutdown gracefully:
import signal

shutdown_event = asyncio.Event()

def signal_handler(sig, frame):
    logger.info("Shutting down gracefully...")
    shutdown_event.set()

signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)

async def main():
    async with fast.run() as agent:
        try:
            await shutdown_event.wait()
        finally:
            logger.info("Cleanup complete")

​
Performance Optimization

​
Streaming Responses

@app.post("/stream")
async def stream_response(req: AskRequest):
    async def generate():
        async for chunk in app.state.agent.stream(req.message):
            yield f"data: {chunk}\n\n"
    
    return StreamingResponse(generate(), media_type="text/event-stream")

​
Caching

from functools import lru_cache
import hashlib

class CachedAgent:
    def __init__(self, agent):
        self.agent = agent
        self.cache = {}
    
    async def send(self, message: str) -> str:
        # Simple hash-based caching
        cache_key = hashlib.md5(message.encode()).hexdigest()
        
        if cache_key in self.cache:
            return self.cache[cache_key]
        
        result = await self.agent.send(message)
        self.cache[cache_key] = result
        return result

​
Connection Pooling

@fast.agent(
    name="pooled_agent",
    servers=["fetch"],
    connection_persistence=True,  # Reuse MCP connections
)
async def main():
    async with fast.run() as agent:
        # Connections are pooled and reused
        await agent.interactive()

​
Next Steps

Configuration

Learn about all configuration options

Basic Examples

Back to basic usage examples

API Reference

Explore the complete API

GitHub

Contribute to Fast Agent on GitHub

Build docs developers (and LLMs) love

Get started for freeTalk to us