Wrapping Internal REST APIs as MCP Tools

The most common mistake when exposing an internal REST API through MCP is a mechanical 1:1 mapping: one tool per endpoint. This preserves the API’s internal structure rather than designing around what an agent actually needs to accomplish, and it forces the model to chain many low-level calls to do anything useful. This lesson works through tool granularity, error translation, and a question that comes up constantly with older internal APIs: how much of the underlying system’s rough edges should the MCP layer smooth over.

Design Tools Around Tasks, Not Endpoints

Consider a customer service REST API with GET /users/{id}, GET /users/{id}/orders, GET /users/{id}/addresses, and GET /users/{id}/payment-methods. Exposing four separate tools means the model must make four sequential calls (and reason about the right order) just to answer “give me this customer’s profile.” A better design consolidates that into one task-shaped tool.

flowchart LR
    subgraph naive["Naive: 1:1 endpoint mapping"]
        M1["Model"] --> E1["get_user"]
        M1 --> E2["get_user_orders"]
        M1 --> E3["get_user_addresses"]
        M1 --> E4["get_user_payment_methods"]
    end

    subgraph better["Task-oriented tool"]
        M2["Model"] --> T["get_customer_profile"]
        T --> E1b["GET /users/{id}"]
        T --> E2b["GET /users/{id}/orders"]
        T --> E3b["GET /users/{id}/addresses"]
    end

    style M1 fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style M2 fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style T fill:#f0fdf9,stroke:#0D9488,color:#0F172A

class CustomerProfile(BaseModel):
    user: dict
    recent_orders: list[dict]
    primary_address: dict | None

@mcp.tool(name="get_customer_profile")
async def get_customer_profile(user_id: str) -> CustomerProfile:
    """
    Fetch a consolidated customer profile: basic info, the 5 most recent
    orders, and primary shipping address. Use this instead of separate
    lookups whenever a task needs a general view of a customer.
    """
    async with httpx.AsyncClient(base_url=INTERNAL_API_BASE) as client:
        user_resp, orders_resp, addr_resp = await asyncio.gather(
            client.get(f"/users/{user_id}"),
            client.get(f"/users/{user_id}/orders", params={"limit": 5}),
            client.get(f"/users/{user_id}/addresses", params={"type": "primary"}),
        )
    return CustomerProfile(
        user=user_resp.json(),
        recent_orders=orders_resp.json()["orders"],
        primary_address=(addr_resp.json() or {}).get("address"),
    )

Keep a small number of narrow, single-purpose tools alongside the consolidated one (e.g. update_user_address for a targeted write) rather than trying to force every operation through a single “do everything” tool, granularity should match the shape of real tasks, not always favor maximal consolidation. A useful heuristic: if a real support agent or salesperson, doing this task by hand, would naturally think of it as “one thing” (checking a customer’s overall status), consolidate; if they would naturally think of it as several distinct actions with independent triggers (looking someone up versus updating their address, which happens far less often and for a different reason), keep them separate.

Handling Partial Failure Across a Consolidated Call

The consolidation above introduces a new failure mode a single-endpoint tool never had: what happens when two of the three underlying calls succeed but one fails? Silently returning a profile with a missing field, with no indication anything went wrong, risks the model treating an absent address as “this customer has no address” rather than “the address lookup failed.” The fix is to surface partial failure explicitly, in the same spirit as the batch partial-success pattern from Lesson 6.

class CustomerProfileResult(BaseModel):
    user: dict
    recent_orders: list[dict] | None
    primary_address: dict | None
    warnings: list[str] = []

@mcp.tool(name="get_customer_profile")
async def get_customer_profile(user_id: str) -> CustomerProfileResult:
    """Consolidated customer profile. Reports which sub-lookups failed, if any, via `warnings`."""
    async with httpx.AsyncClient(base_url=INTERNAL_API_BASE) as client:
        user_resp = await client.get(f"/users/{user_id}")
        user_resp.raise_for_status()  # This one is load-bearing; fail the whole tool if it fails

        warnings = []
        try:
            orders_resp = await client.get(f"/users/{user_id}/orders", params={"limit": 5})
            orders_resp.raise_for_status()
            recent_orders = orders_resp.json()["orders"]
        except httpx.HTTPError:
            recent_orders = None
            warnings.append("Could not fetch recent orders, order service may be degraded")

        try:
            addr_resp = await client.get(f"/users/{user_id}/addresses", params={"type": "primary"})
            addr_resp.raise_for_status()
            primary_address = (addr_resp.json() or {}).get("address")
        except httpx.HTTPError:
            primary_address = None
            warnings.append("Could not fetch primary address, address service may be degraded")

    return CustomerProfileResult(
        user=user_resp.json(), recent_orders=recent_orders,
        primary_address=primary_address, warnings=warnings,
    )

Note the distinction between the core user_resp call (whose failure should fail the entire tool, there is no meaningful “customer profile” without basic user data) and the two secondary calls (whose individual failure degrades the result but should not block the whole response). Deciding which calls are load-bearing versus best-effort is a design judgment specific to each tool, not a mechanical rule, but making that judgment explicit in code, rather than letting any single failed await silently propagate as an unhandled exception, is what keeps consolidated tools trustworthy.

Translating REST Errors into MCP Errors

Raw HTTP status codes are meaningless to a model unless translated into actionable, structured tool errors.

@mcp.tool(name="update_user_address")
async def update_user_address(user_id: str, address: AddressInput) -> dict:
    """Update a customer's primary shipping address."""
    try:
        resp = await internal_api_client.put(f"/users/{user_id}/addresses/primary", json=address.model_dump())
        resp.raise_for_status()
        return {"success": True, "address": resp.json()}
    except httpx.HTTPStatusError as e:
        if e.response.status_code == 404:
            return {"isError": True, "retryable": False, "message": f"No user found with id '{user_id}'"}
        if e.response.status_code == 429:
            retry_after = e.response.headers.get("Retry-After", "60")
            return {"isError": True, "retryable": True, "message": f"Rate limited, retry after {retry_after}s"}
        if e.response.status_code == 422:
            return {"isError": True, "retryable": False, "message": f"Invalid address: {e.response.json().get('detail')}"}
        raise  # Unexpected 5xx: let it surface as a genuine execution failure

Normalizing Legacy API Inconsistency at the Wrapping Layer

Internal REST APIs, especially older ones accumulated over years across different subsystems and teams, are rarely internally consistent: date formats vary, error response shapes vary, pagination conventions vary between endpoints written by different people at different times. A natural question when wrapping such an API is whether the MCP tool should faithfully mirror these inconsistencies or paper over them. The strong recommendation is to normalize at the wrapping layer: present one consistent date format, one consistent error shape, and one consistent pagination convention across every tool, regardless of which legacy quirk produced the underlying raw response.

from datetime import datetime

def normalize_date(raw_value) -> str:
    """Every tool in this server returns ISO 8601 dates, regardless of
    whether the underlying legacy endpoint returned a Unix timestamp,
    an ISO string, or a locale-formatted date string."""
    if isinstance(raw_value, (int, float)):
        return datetime.utcfromtimestamp(raw_value).isoformat() + "Z"
    if isinstance(raw_value, str) and "/" in raw_value:  # legacy MM/DD/YYYY subsystem
        return datetime.strptime(raw_value, "%m/%d/%Y").isoformat() + "Z"
    return raw_value  # already ISO 8601

This is one of the highest-value, lowest-risk changes an MCP wrapping layer can make: it costs nothing to the underlying system (no changes to the legacy API itself are required) and it removes an entire class of model confusion where output that looks superficially similar (a date, an error) actually varies unpredictably in format depending on which internal code path happened to generate it. The next lesson applies this same design discipline, granularity, partial-failure handling, and normalization, to a specific, high-value enterprise category: database and data warehouse access.