Tool Integration Patterns

By the end of this lesson, you'll understand how agents connect to external tools via the Model Context Protocol (MCP), choose the right architecture pattern for your deployment, and apply production-grade best practices for reliability and security.

The Model Context Protocol (MCP)

Anthropic introduced the Model Context Protocol in November 2024 as an open standard for connecting AI models to external data sources and tools. MCP solves the "M×N integration problem" — without it, every model needs a custom connector for every tool. With MCP, tools expose a standard interface that any compliant model can consume.

MCP follows a client-server architecture. The AI application (client) connects to MCP servers, each of which exposes a set of tools. The server declares what it can do, and the model decides when to call each tool based on the user's task.

Server Pattern: Declaring and Handling Tools

An MCP server uses two core decorators to expose tools:

from mcp.server import Server
from mcp.types import Tool, TextContent

server = Server("my-tool-server")

@server.list_tools()
async def list_tools() -> list[Tool]:
    """Declare available tools and their input schemas."""
    return [
        Tool(
            name="search_documents",
            description="Search internal docs by keyword",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Search query"
                    },
                    "limit": {
                        "type": "integer",
                        "description": "Max results",
                        "default": 10
                    }
                },
                "required": ["query"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """Execute the named tool with provided arguments."""
    if name == "search_documents":
        results = await search_index(
            query=arguments["query"],
            limit=arguments.get("limit", 10)
        )
        return [TextContent(
            type="text",
            text=format_results(results)
        )]
    raise ValueError(f"Unknown tool: {name}")

The @server.list_tools() decorator registers a function that returns tool definitions — name, description, and JSON Schema for inputs. The model uses these descriptions to decide when a tool is relevant. The @server.call_tool() decorator registers the handler that actually executes tool logic when the model invokes it.

Why MCP Matters at Scale

Without MCP, integrating 5 tools across 3 agent frameworks requires 15 custom connectors. With MCP, you write 5 servers and every framework connects to all of them. This is the same insight that made USB universal — a standard interface eliminates combinatorial explosion.

Three Architecture Patterns

The AWS governance guide identifies three patterns for how agents access tools. Each makes different tradeoffs between latency, governance, and isolation.

Pattern 1: In-Runtime Tools

Tools run in the same process as the agent. The agent calls tool functions directly — no network hop, no serialization overhead.

When to use: Prototypes, single-developer tools, latency-critical paths where sub-millisecond tool response matters, tools with no external dependencies.

Example: A code formatting tool that runs a linter in-process, or a calculator that evaluates expressions locally.

Pattern 2: Direct Access (Remote Tools)

The agent calls external services directly — APIs, databases, or MCP servers running on separate infrastructure. Each tool integration is a point-to-point connection.

When to use: Small teams, moderate tool counts (under 10), tools owned by the same team, environments where centralized governance isn't required.

Example: An agent that directly calls a Jira API, a GitHub API, and a Slack webhook — each with its own auth credentials configured in the agent.

Pattern 3: Tools Gateway

A centralized gateway sits between agents and tools. All tool calls route through it, enabling consistent auth, rate limiting, logging, and policy enforcement across every integration.

When to use: Enterprise deployments, multi-team environments, regulated industries, any scenario requiring centralized audit trails or credential rotation.

Example: A platform team runs the gateway. Agent teams register which tools they need. The gateway handles OAuth token refresh, enforces per-team rate limits, and logs every call for compliance.

Tradeoff Comparison

Production Best Practices

1. Environment Variables for Secrets

Never hardcode API keys, tokens, or credentials. Use environment variables or a secrets manager:

import os
from dataclasses import dataclass

@dataclass(frozen=True)
class ToolConfig:
    api_key: str
    base_url: str
    timeout_seconds: int = 30

    @classmethod
    def from_env(cls) -> "ToolConfig":
        api_key = os.environ.get("TOOL_API_KEY")
        if not api_key:
            raise RuntimeError(
                "TOOL_API_KEY not set. "
                "See docs/setup.md for configuration."
            )
        return cls(
            api_key=api_key,
            base_url=os.environ.get(
                "TOOL_BASE_URL",
                "https://api.example.com"
            ),
            timeout_seconds=int(
                os.environ.get("TOOL_TIMEOUT", "30")
            )
        )

2. Retry with Exponential Backoff

Network calls fail. Retries with exponential backoff and jitter prevent thundering herds:

import asyncio
import random

async def call_with_retry(
    fn,
    max_attempts: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 30.0
):
    """Retry with exponential backoff and jitter."""
    for attempt in range(max_attempts):
        try:
            return await fn()
        except (TimeoutError, ConnectionError) as e:
            if attempt == max_attempts - 1:
                raise
            delay = min(
                base_delay * (2 ** attempt),
                max_delay
            )
            # Add jitter: 0.5x to 1.5x the delay
            jittered = delay * (0.5 + random.random())
            await asyncio.sleep(jittered)

The jitter prevents synchronized retries when multiple agents hit the same tool simultaneously. Without jitter, all failed requests retry at the same moment, causing repeated failures.

3. Timeout Handling

Every external call needs a timeout. An unresponsive tool should not block the agent indefinitely:

import asyncio
import httpx

async def call_tool_with_timeout(
    url: str,
    payload: dict,
    timeout: float = 10.0
) -> dict:
    """Call external tool with strict timeout."""
    async with httpx.AsyncClient(
        timeout=httpx.Timeout(timeout)
    ) as client:
        try:
            response = await client.post(url, json=payload)
            response.raise_for_status()
            return response.json()
        except httpx.TimeoutException:
            return {
                "error": "Tool timed out",
                "fallback": "Proceeding without tool result"
            }
        except httpx.HTTPStatusError as e:
            return {
                "error": f"Tool returned {e.response.status_code}",
                "retryable": e.response.status_code >= 500
            }

4. Input Validation

Validate tool inputs before sending them to external services. The model can produce malformed arguments:

from pydantic import BaseModel, Field, ValidationError

class SearchInput(BaseModel):
    query: str = Field(min_length=1, max_length=500)
    limit: int = Field(default=10, ge=1, le=100)
    filters: dict = Field(default_factory=dict)

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "search_documents":
        try:
            validated = SearchInput(**arguments)
        except ValidationError as e:
            return [TextContent(
                type="text",
                text=f"Invalid input: {e.errors()[0]['msg']}"
            )]
        return await execute_search(validated)

Validation catches issues like empty queries, negative limits, or oversized payloads before they reach your backend. Return clear error messages so the model can self-correct.

5. Response Caching

Cache tool results when the underlying data doesn't change between calls:

from functools import lru_cache
from datetime import datetime, timedelta

class ToolCache:
    def __init__(self, ttl_seconds: int = 300):
        self._cache: dict[str, tuple[datetime, any]] = {}
        self._ttl = timedelta(seconds=ttl_seconds)

    def get(self, key: str):
        if key in self._cache:
            timestamp, value = self._cache[key]
            if datetime.now() - timestamp < self._ttl:
                return value
            del self._cache[key]
        return None

    def set(self, key: str, value):
        self._cache[key] = (datetime.now(), value)

# Usage in tool handler
cache = ToolCache(ttl_seconds=60)

async def search_with_cache(query: str, limit: int):
    cache_key = f"search:{query}:{limit}"
    cached = cache.get(cache_key)
    if cached is not None:
        return cached
    result = await execute_search(query, limit)
    cache.set(cache_key, result)
    return result

Caching is especially valuable for tools the model calls repeatedly within one conversation — looking up the same documentation, checking the same config, or querying the same database state.

Putting It Together: A Complete MCP Server

Here's how these practices combine into a production-ready tool server:

import os
import asyncio
from mcp.server import Server
from mcp.types import Tool, TextContent
from pydantic import BaseModel, Field

server = Server("production-tool-server")

# Best practice: config from environment
API_KEY = os.environ["SEARCH_API_KEY"]
TIMEOUT = int(os.environ.get("SEARCH_TIMEOUT", "10"))

# Best practice: input validation
class SearchParams(BaseModel):
    query: str = Field(min_length=1, max_length=500)
    limit: int = Field(default=10, ge=1, le=100)

# Best practice: caching
_cache: dict[str, tuple[float, list]] = {}
CACHE_TTL = 60  # seconds

@server.list_tools()
async def list_tools() -> list[Tool]:
    return [Tool(
        name="search_docs",
        description=(
            "Search internal documentation. Returns title, "
            "snippet, and URL for each match. Use when the "
            "user asks about internal processes or policies."
        ),
        inputSchema=SearchParams.model_json_schema()
    )]

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name != "search_docs":
        raise ValueError(f"Unknown tool: {name}")

    # Validate
    params = SearchParams(**arguments)
    cache_key = f"{params.query}:{params.limit}"

    # Check cache
    import time
    if cache_key in _cache:
        ts, result = _cache[cache_key]
        if time.time() - ts < CACHE_TTL:
            return [TextContent(type="text", text=result)]

    # Call with retry + timeout
    result = await call_with_retry(
        lambda: search_api(params, timeout=TIMEOUT)
    )

    # Cache and return
    formatted = format_results(result)
    _cache[cache_key] = (time.time(), formatted)
    return [TextContent(type="text", text=formatted)]

Verify Your Understanding

Apply It

Take one of your existing tool integrations — a direct API call, a database query, a file system operation — and refactor it into an MCP server. Add input validation, a 10-second timeout, and a 60-second cache. Then ask: does this tool belong in-runtime, direct access, or behind a gateway?