Keyboard shortcuts

Press ← or → to navigate between chapters

Press S or / to search in the book

Press ? to show this help

Press Esc to hide this help

Chapter 9: API Integration Patterns

Wrapping an API with MCP sounds straightforward until you try it. Existing APIs were designed for human developers reading documentation, not for models deciding what to call in the middle of a conversation. The endpoint names are cryptic. The payloads are bloated. The error codes assume you have the Swagger page open in another tab.

The job of an MCP server that fronts an API is not to mirror that API. It is to translate it into a smaller, clearer, safer capability surface—one where the model can express user intent without needing to understand the plumbing underneath. The server owns everything the model shouldn’t have to know: credentials, rate limits, error codes, and which of the API’s hundred fields answer the question.

The Rule Under Every Pattern: Return What the Question Needs

API wrappers frequently over-return. They send the entire downstream payload back to the client because “maybe the model will need it.” Usually it does not.

Every field in a tool response costs tokens. A GitHub pull request object has over 100 fields. Twenty of them in full come to roughly 400 KB of JSON, on the order of 100,000 tokens, spent on merge_commit_sha, _links.statuses.href and head.repo.allow_squash_merge. Number, title, author and date for the same 20 PRs come to about 3 KB, under 1,000 tokens, and those four fields answer most PR questions.

Shape every response:

  • Return the fields needed for the task, plus identifiers for follow-up. The PR number lets the model ask for details.
  • Derive status the model can reason about. A boolean mergeable beats the raw merge state machine.
  • Offer a detail tool separately. A get_pr_details tool returns the full object on demand instead of in every list.

If a list response carries more than 10 fields per item, you are probably over-returning. Each of the four patterns below applies this rule to a different kind of downstream API.

Pattern 1: The Legacy Wrapper

Legacy APIs accumulate scar tissue. SOAP envelopes. Field names like acct_nmbr or txn_dt. Multi-step rituals where you authenticate, fetch a session token, pass it as a header to a second endpoint, then parse XML to find the one field you actually needed. These APIs work—they have worked for years—but they were never designed to be described to a language model.

A legacy wrapper MCP server absorbs that complexity. The model sees clean tool names, readable parameters, and structured responses. The server handles the ugly translation layer.

Here is a concrete example. Suppose you need to wrap a legacy inventory service that exposes a SOAP endpoint. The downstream call requires an XML envelope, returns nested XML, and uses field names that only make sense to the team that built it in 2009.

from xml.sax.saxutils import escape

import httpx
import xmltodict
from mcp.server.mcpserver import MCPServer
from mcp.server.mcpserver.exceptions import ToolError

mcp = MCPServer("Inventory Server")

INVENTORY_URL = "https://internal.example.com/soap/inventory"

SOAP_TEMPLATE = """<?xml version="1.0" encoding="UTF-8"?>
<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"
                  xmlns:inv="http://example.com/inventory">
  <soapenv:Body>
    <inv:GetStockLevel>
      <inv:SKU>{sku}</inv:SKU>
      <inv:LocCd>{location_code}</inv:LocCd>
    </inv:GetStockLevel>
  </soapenv:Body>
</soapenv:Envelope>"""

# Human-readable warehouse names → legacy location codes
LOCATION_MAP = {
    "us-east-1": "WHSE_USE1",
    "us-west-2": "WHSE_USW2",
    "eu-central": "WHSE_EUC1",
}


@mcp.tool()
async def check_stock(
    sku: str,
    warehouse: str = "us-east-1",
) -> dict:
    """Check current stock level for a product at a specific warehouse.

    Returns the available quantity, reserved quantity,
    and whether the item is below its reorder threshold.
    """
    location_code = LOCATION_MAP.get(warehouse)
    if not location_code:
        raise ToolError(f"Unknown warehouse: {warehouse}. "
                        f"Valid options: {', '.join(LOCATION_MAP)}")

    # SOAP has no parameters: escape model-supplied text before it
    # goes anywhere near the envelope
    body = SOAP_TEMPLATE.format(sku=escape(sku), location_code=location_code)
    headers = {"Content-Type": "text/xml; charset=utf-8",
               "SOAPAction": "GetStockLevel"}

    async with httpx.AsyncClient() as client:
        resp = await client.post(INVENTORY_URL, content=body,
                                 headers=headers, timeout=10.0)

    if resp.status_code != 200:
        raise ToolError(f"Inventory service returned {resp.status_code}. "
                        "The service may be temporarily unavailable.")

    parsed = xmltodict.parse(resp.text)
    try:
        result = (parsed["soapenv:Envelope"]["soapenv:Body"]
                  ["inv:GetStockLevelResponse"]["inv:Result"])
    except KeyError:
        raise ToolError("Unexpected response structure from inventory service.")

    available = int(result.get("inv:AvailQty", 0))
    reserved = int(result.get("inv:RsvdQty", 0))
    reorder_point = int(result.get("inv:ReordPt", 0))

    return {
        "sku": sku,
        "warehouse": warehouse,
        "available": available,
        "reserved": reserved,
        "below_reorder_threshold": available < reorder_point,
    }

Notice what the wrapper does:

  • Normalizes naming. The model never sees inv:AvailQty or WHSE_USE1. It sees available and us-east-1.
  • Collapses multi-step rituals. One tool call replaces the SOAP envelope construction, header setup, XML parsing, and field extraction.
  • Translates errors into actionable text. Instead of a raw 500 or a stack trace, the model gets a sentence it can relay to the user or use to decide its next step.
  • Escapes inputs. SOAP is not parameterized, so the wrapper has to be. An unescaped sku of </inv:SKU><inv:LocCd>WHSE_EUC1</inv:LocCd><inv:SKU> rewrites the request. This is SQL injection with angle brackets.
  • Hides transport details. The model does not know or care that the backend speaks SOAP over HTTP. It called a tool and got JSON back.

The general principle: if the downstream API is awkward, that is an argument for mediation. It is not a reason to export the awkwardness upstream.

When OpenAPI Automation Works

If the API you are wrapping has a clean OpenAPI specification, you can generate tools from it. The official SDK doesn’t do this. FastMCP.from_openapi() is a feature of the standalone fastmcp package, a separate project from the official mcp SDK used elsewhere in this book. You provide an httpx.AsyncClient configured with the base URL and credentials, plus the spec, and it creates one tool per endpoint.

import os

import httpx
from fastmcp import FastMCP  # standalone fastmcp package, not the official SDK

client = httpx.AsyncClient(
    base_url="https://api.example.com/v2",
    headers={"Authorization": f"Bearer {os.environ['API_TOKEN']}"},
)

mcp = FastMCP.from_openapi(
    openapi_spec=spec_dict,
    client=client,
    name="Example API Server",
)

This approach works well when the API is already well-designed—clean endpoint names, reasonable parameter counts, useful descriptions. It breaks down when the API surface is large (dozens of endpoints become dozens of tools, overwhelming the model’s context window) or when the endpoint design leaks internal concerns. Use automatic generation as a starting point, then prune and reshape.

A useful heuristic: if you find yourself wanting to rename more than a third of the generated tools, write the wrapper by hand. The auto-generated server was saving you time, not adding clarity.

Pattern 2: The Auth Proxy

The MCP server should be the place where credentials are attached, refreshed, and rotated. The model should never be asked for secrets and should never see raw tokens in tool responses unless the workflow explicitly demands it and the security model supports it.

API keys passed as tool parameters and tokens leaked in verbose error messages were common in early MCP servers. In a multi-turn conversation the full history is visible to the model and often logged, so every one of those is a credential exposure.

Runtime Secret Injection

The safer pattern is runtime injection: credentials are loaded from environment variables or a secret manager at server startup and attached to outgoing requests inside the tool handler. They never appear in tool schemas, tool descriptions, or tool responses.

import os

import httpx
from mcp.server.mcpserver import MCPServer
from mcp.server.mcpserver.exceptions import ToolError

mcp = MCPServer("GitHub Proxy")

# Credentials loaded once at startup—never exposed to the model
GITHUB_TOKEN = os.environ["GITHUB_TOKEN"]
GITHUB_HEADERS = {
    "Authorization": f"Bearer {GITHUB_TOKEN}",
    "Accept": "application/vnd.github.v3+json",
}


@mcp.tool()
async def list_open_pull_requests(
    owner: str,
    repo: str,
    limit: int = 10,
) -> dict:
    """List open pull requests for a GitHub repository.

    Returns title, number, author, and creation date for each PR.
    """
    url = f"https://api.github.com/repos/{owner}/{repo}/pulls"
    params = {"state": "open", "per_page": min(limit, 100)}

    async with httpx.AsyncClient() as client:
        resp = await client.get(url, headers=GITHUB_HEADERS,
                                params=params, timeout=10.0)

    if resp.status_code == 401:
        raise ToolError("GitHub authentication failed. "
                        "The server's credentials may need rotation.")
    if resp.status_code == 403:
        # GitHub uses 403 for both exhausted quota and missing permission
        if resp.headers.get("X-RateLimit-Remaining") == "0":
            reset = resp.headers.get("X-RateLimit-Reset", "unknown")
            raise ToolError("GitHub rate limit exhausted. "
                            f"Quota resets at epoch {reset}; don't retry before then.")
        if "Retry-After" in resp.headers:  # secondary (burst) limit
            raise ToolError("GitHub secondary rate limit hit. Retry after "
                            f"{resp.headers['Retry-After']} seconds.")
        raise ToolError(f"The server's GitHub token can't read {owner}/{repo}. "
                        "Retrying won't help; ask an admin to grant access.")
    if resp.status_code != 200:
        raise ToolError(f"GitHub API returned {resp.status_code}.")

    prs = resp.json()
    return {
        "repository": f"{owner}/{repo}",
        "open_prs": [
            {
                "number": pr["number"],
                "title": pr["title"],
                "author": pr["user"]["login"],
                "created": pr["created_at"][:10],
            }
            for pr in prs
        ],
    }

The key detail: the Authorization header is injected in the handler, not described in the tool schema. The model knows it can list pull requests. It does not know how authentication works, and it should not.

The 403 branch shows why status codes need translating rather than forwarding. GitHub returns 403 both when quota runs out and when the token lacks access to the repository. Reporting both as “rate limit hit” sends the model into a wait-and-retry loop that can never succeed. X-RateLimit-Remaining: 0 or a Retry-After header (GitHub’s secondary, burst limit) is what tells them apart.

Credential Rotation Without Restart

For long-running remote MCP servers, credentials expire. If you load the token once at import time and never refresh it, the server silently breaks when the token expires hours or days later. With an OAuth client-credentials grant, the server fetches its own token and caches it for as long as the identity provider says it’s valid:

import os
import time

import httpx

_token = {"value": None, "expires_at": 0.0}


async def get_api_token() -> str:
    """Return a valid access token, fetching a new one near expiry."""
    if _token["value"] and time.time() < _token["expires_at"] - 60:
        return _token["value"]

    async with httpx.AsyncClient() as client:
        resp = await client.post(
            os.environ["TOKEN_URL"],
            data={"grant_type": "client_credentials",
                  "scope": "inventory:read"},
            auth=(os.environ["CLIENT_ID"], os.environ["CLIENT_SECRET"]),
            timeout=10.0,
        )
    resp.raise_for_status()
    body = resp.json()
    _token["value"] = body["access_token"]
    _token["expires_at"] = time.time() + body["expires_in"]
    return _token["value"]

Each tool handler calls get_api_token() on every invocation rather than closing over a value read at startup. Expiry comes from the provider’s expires_in, not an assumption. The 60-second margin keeps a token from expiring between the check and the downstream call.

What Not to Do

Avoid these patterns, all of which have appeared in real MCP servers:

  • API key as a tool parameter. def query_database(api_key: str, query: str) — the model will echo the key in conversation history.
  • Credentials in error messages. raise ToolError(f"Auth failed with token {token[:8]}...") — partial tokens are still useful to an attacker.
  • Shared credentials across tenants. In multi-tenant deployments, each user’s requests should carry that user’s scoped credential, not a single god-token. See Chapter 18 for the full identity model.

A good rule: the model should express intent, not credential strategy.

Pattern 3: The Rate Limiter

Models are capable of fast, repetitive tool usage. A model asked to “check the status of all 200 tickets” will happily fire 200 sequential tool calls as fast as the transport allows. If each call hits an external API with a 100-request-per-minute limit, your downstream service is overwhelmed in seconds, and your API key may be suspended.

Rate limiting in an MCP server is not optional. It is a safety requirement for any server that proxies to external services.

Where to Limit

Limit per downstream service first, because that’s the quota you can lose. Then limit per caller, so one user can’t spend everyone’s share. Per-tool and global limits are refinements you add when a measurement asks for them.

Token Bucket Implementation

The token bucket algorithm is the standard choice for MCP rate limiting because it allows short bursts while enforcing an average rate. Here is a minimal implementation. acquire takes a cost, so expensive calls can draw more than one token:

import asyncio
import time


class TokenBucket:
    """Token bucket rate limiter with weighted acquisition."""

    def __init__(self, rate: float, burst: int):
        self.rate = rate        # tokens per second
        self.burst = burst      # max tokens (bucket size)
        self.tokens = burst
        self.last_refill = time.monotonic()
        self._lock = asyncio.Lock()

    async def acquire(self, cost: int = 1) -> bool:
        async with self._lock:
            now = time.monotonic()
            elapsed = now - self.last_refill
            self.tokens = min(self.burst,
                              self.tokens + elapsed * self.rate)
            self.last_refill = now

            if self.tokens >= cost:
                self.tokens -= cost
                return True
            return False

    def retry_after(self, cost: int = 1) -> float:
        """Seconds until `cost` tokens are available."""
        return max(0.0, (cost - self.tokens) / self.rate)


# One bucket per (tool_name, caller) pair
_buckets: dict[tuple[str, str], TokenBucket] = {}


def get_bucket(tool_name: str, caller: str,
               rate: float = 0.5, burst: int = 5) -> TokenBucket:
    key = (tool_name, caller)
    if key not in _buckets:
        _buckets[key] = TokenBucket(rate=rate, burst=burst)
    return _buckets[key]

Backpressure as Interface Design

The critical difference between good and bad rate limiting is what happens when the limit is reached. If the server returns a generic error—“rate limit exceeded”—the model has no information to guide its next step. It may retry immediately, making the problem worse.

A better response includes three things: what happened, when to retry, and what alternative the model might consider. The key the limit is counted against matters just as much. It has to be the authenticated caller:

from mcp.server.auth.middleware.auth_context import get_access_token


def caller_key() -> str:
    """The authenticated principal for this request, from the verified token."""
    token = get_access_token()
    if token is None:
        raise ToolError("This tool requires an authenticated caller.")
    return f"{token.client_id}:{token.subject}"


@mcp.tool()
async def search_tickets(query: str) -> dict:
    """Search support tickets by keyword."""
    bucket = get_bucket("search_tickets", caller_key(), rate=0.5, burst=5)

    if not await bucket.acquire():
        retry = round(bucket.retry_after(), 1)
        raise ToolError(
            f"Rate limit reached for ticket search. Retry after {retry}s. "
            "Several searches ran in quick succession; a more specific "
            "query would need fewer calls."
        )

    # ... actual search logic ...

This is backpressure as interface design. The model can read the retry time and wait, or it can read the suggestion and change strategy. Either outcome is better than a blind retry loop.

The caller comes from the request context: the SDK exposes the token that the server’s auth layer already verified. It never comes from a tool argument. A limiter that takes user_id: str = "default" as a parameter gets default from the model, puts every user in one global bucket, and lets any caller exhaust someone else’s quota by naming them. Never key a limit on something the model supplies. It will supply default, or someone else’s id.

Cost-Aware Limiting

Not all tool calls are equal. A call that triggers a simple database lookup costs far less than one that invokes a third-party API with per-request pricing. Weight each tool by what it spends:

TOOL_COSTS = {
    "quick_lookup": 1,       # cheap, local
    "full_analysis": 10,     # calls paid API
    "generate_report": 25,   # calls multiple paid APIs
}

# inside a handler, with a shared per-caller budget bucket (burst=50)
if not await budget.acquire(cost=TOOL_COSTS["full_analysis"]):
    ...

With a burst of 50, a model can run two generate_report calls or fifty quick_lookup calls before it has to wait. That stops one conversation from spending the day’s API budget on the expensive tool.

Pattern 4: The Aggregator

Some of the best MCP servers exist not to wrap a single API but to combine multiple APIs into a capability that no single service provides. Correlating pull requests with ticket status. Joining CRM data with support history. Combining deployment logs with observability signals.

The model gets to ask “what is the status of this feature?” rather than “query GitHub, then query Jira, then correlate by branch name.” The difficulty is provenance and failure handling.

A Concrete Aggregator

Here is a server that combines GitHub and a project tracker to answer the question “what work is in flight?”:

import asyncio
import os
import re

import httpx
from mcp.server.mcpserver import MCPServer

mcp = MCPServer("Work Status Aggregator")

GITHUB_TOKEN = os.environ["GITHUB_TOKEN"]
TRACKER_TOKEN = os.environ["TRACKER_TOKEN"]


async def fetch_github_prs(github_repo: str) -> list[dict]:
    """Fetch open PRs from GitHub."""
    url = f"https://api.github.com/repos/{github_repo}/pulls"
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            url,
            headers={"Authorization": f"Bearer {GITHUB_TOKEN}"},
            params={"state": "open", "per_page": 50},
            timeout=10.0,
        )
    if resp.status_code != 200:
        raise RuntimeError(f"GitHub returned {resp.status_code}")
    return resp.json()


async def fetch_tracker_tickets(project_id: str) -> list[dict]:
    """Fetch active tickets from the project tracker."""
    url = f"https://tracker.example.com/api/projects/{project_id}/tickets"
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            url,
            headers={"Authorization": f"Bearer {TRACKER_TOKEN}"},
            params={"status": "active"},
            timeout=10.0,
        )
    if resp.status_code != 200:
        raise RuntimeError(f"Tracker returned {resp.status_code}")
    return resp.json()


def branch_for(ticket_id: str, branches: list[str]) -> str | None:
    """Branch named for this ticket: PROJ-12 matches feat/PROJ-12-x, not PROJ-123."""
    pattern = re.compile(rf"(^|/){re.escape(ticket_id)}(-|$)", re.IGNORECASE)
    return next((b for b in branches if pattern.search(b)), None)


@mcp.tool()
async def work_in_flight(github_repo: str, project_id: str) -> dict:
    """Show active work items with their PR and ticket status.

    github_repo is "owner/repo". Correlates GitHub pull requests with
    tracker tickets by branch name. Every item says which source it
    came from; partial_failure lists any source that didn't answer.
    """
    prs, tickets = await asyncio.gather(
        fetch_github_prs(github_repo),
        fetch_tracker_tickets(project_id),
        return_exceptions=True,
    )
    errors = []
    if isinstance(prs, Exception):
        errors.append(f"GitHub: {prs}")
        prs = []
    if isinstance(tickets, Exception):
        errors.append(f"Tracker: {tickets}")
        tickets = []

    pr_by_branch = {
        pr["head"]["ref"]: {
            "source": "github",
            "number": pr["number"],
            "title": pr["title"],
            "author": pr["user"]["login"],
            "url": pr["html_url"],
        }
        for pr in prs
    }

    linked, used = [], set()
    for ticket in tickets:
        branch = branch_for(ticket["id"], list(pr_by_branch))
        if branch:
            used.add(branch)
        linked.append({
            "source": "tracker",
            "ticket_id": ticket["id"],
            "ticket_title": ticket["title"],
            "ticket_status": ticket["status"],
            "pr": pr_by_branch.get(branch),
        })

    response = {
        "linked_items": linked,
        "orphan_prs": [pr for b, pr in pr_by_branch.items() if b not in used],
    }
    if errors:
        response["partial_failure"] = errors
        response["note"] = ("Some data sources were unavailable. "
                            "Results may be incomplete.")
    return response

Several things matter in this implementation:

Partial failure tolerance. Both sources are fetched concurrently, so the call takes as long as the slower source, not the sum of the two. return_exceptions=True turns a failed fetch into a value instead of cancelling the other one. If GitHub is down but the tracker responds, the model still gets the tickets, and partial_failure tells it exactly what is missing. The model can say “I have ticket status but couldn’t reach GitHub” instead of failing entirely. A partial result isn’t a tool error. It’s a successful call that says what it lacks.

Provenance tracking. Every item carries a source field. When the model presents results, it can say where each fact came from, and a user who doubts a status knows which system to check.

Correlation logic lives in the server. The branch-name-to-ticket matching convention is domain knowledge that should not be pushed to the model. The model would need to be told the convention, then execute the join in its reasoning—wasting tokens on mechanical work the server can do deterministically. The server can also get the join right. A substring test ("proj-12" in branch.lower()) links ticket PROJ-12 to feat/PROJ-123-billing. branch_for anchors the ID between a path separator and a hyphen or the end of the name.

Do not merge everything into one giant tool. Aggregate around a clear question. work_in_flight answers “what is in flight?” If the user wants just PRs or just tickets, expose separate tools for those. The aggregator is for the cross-cutting question that neither source answers alone.

Failure Modes

API integration servers have a specific failure profile that differs from servers that wrap local resources. These are the failures you will encounter in production.

Downstream timeout cascading. Your MCP server sets a 30-second timeout. The downstream API is slow but not dead—it responds in 25 seconds. The model makes three calls in one turn. The user watches a spinner for over a minute, and a client with a 60-second tool timeout abandons the call before the answer arrives. The model then reports a failure for work that actually succeeded. Set aggressive timeouts (5–10 seconds for most API calls) and return partial results or clear failure messages rather than waiting.

Credential expiry under load. The server starts fine. An hour later, the OAuth token expires. The next tool call returns a 401. If the server logs the error but returns a generic “something went wrong” to the model, the model cannot help the user. Return “authentication failed—server credentials may need rotation” so the model can relay actionable information.

Rate limit ambiguity. The downstream API returns 429, but your server catches it and returns a generic error. The model retries. The downstream API returns 429 again. The model retries again. You have created a retry amplification loop. Always translate 429s into explicit “rate limited, wait N seconds” responses. The mirror image is a permission error reported as a rate limit, which sends the model waiting for something that will never change.

Spoofable limit keys. A limiter keyed on a tool argument collapses to one bucket when the model fills in the default, and lets any caller drain someone else’s bucket. Key limits on the verified principal.

Substring correlation. An aggregator that joins on id in branch links PROJ-12 to PROJ-123’s pull request and reports it as done. The output looks authoritative and is wrong, and nothing errors. Anchor every join key.

Schema drift. The downstream API changes a field name or removes a field. Your server’s parsing code breaks. Unlike a human developer who would read the changelog, your MCP server just starts returning errors. Defensive parsing—using .get() with defaults rather than direct key access, and validating response structure before extracting fields—reduces the blast radius of upstream changes.

Aggregator partial failure masking. An aggregator tool calls three APIs. One fails. The server catches the exception and returns results from the other two without mentioning the failure. The user sees a result that looks complete but is missing a third of the data. Always surface partial failures explicitly.

Implementation Checklist

Before shipping an API-facing MCP server, verify each of these:

  1. Tool names describe user intent. check_stock beats POST /soap/inventory/GetStockLevel.
  2. No schema, description, response or error message contains a credential. Grep the transcript of a test session for your token prefix.
  3. Tokens come from a fetch that honours expires_in. A test with a 5-second token lifetime passes across a minute of calls.
  4. Every model-supplied string that reaches a non-JSON format is escaped. XML, SQL, shell and URL paths all count.
  5. Every outgoing HTTP call has an explicit timeout of 10 seconds or less.
  6. Limits are keyed on the authenticated caller and weighted by cost. No limiter reads identity from a tool argument.
  7. List responses have a default and a maximum size, and say when more results exist.

What Comes Next

Every pattern in this chapter keeps a sharp edge on the server’s side of the line: the credential, the quota, the join, the escape. Each tool call still stands alone. Chapter 10 takes on work that doesn’t fit in one call, where the server has to remember what happened, wait for a human, or keep a job running after the call returns.