How I Use FastAPI and FastMCP Across My AI Projects

I use FastAPI and FastMCP in two projects, RoboShop AI and MOKS, across four services. They are easy to describe as a pair, but the code says otherwise: no service I have uses both. FastAPI answers people and internal services. FastMCP answers AI clients. Where the two meet, they meet over plain HTTP between separate pods.

This article walks through what each package does in that code, where the design holds up, and where it falls short. Everything below was checked against RoboShop AI at c81e85a and MOKS at ab99cf7.

Where each package runs

Just as useful is what is not on that list. The RoboShop moks shopping assistant runs on Waitress, a WSGI server. The other RoboShop services use neither package. The MOKS portal has no MCP code at all, even though it has a tool layer. This site is Next.js.

An HTTP API and an agent tool are different contracts

Both projects needed the same kind of capability: let something outside a service ask it to do work, safely. But the two callers differ.

An HTTP API is written for a client I control. The browser code knows which endpoint to call and when. If the contract changes, I change both sides.

An agent tool is written for a caller that picks at run time, from a description. An MCP client lists the tools a server offers, passes their names, descriptions and JSON Schemas to a language model, and the model chooses. So a tool definition is an API contract and a prompt at once, and I can't review every decision made with it.

That difference explains the split. FastAPI holds the application boundary, and FastMCP publishes a small, described part of it to callers I don't control.

FastAPI in practice

I use FastAPI in two very different ways. MOKS is conventional. The RoboShop gateway is close to a transport shell.

MOKS: routers, models and dependencies

moks-api is the clearest example. create_app() in backend/moks/main.py mounts 20 routers under one prefix, which produces 142 paths and 161 operations in the OpenAPI schema. Every API path it does not recognise hits a catch-all that returns a JSON 404, so a mistyped API call can't fall through to the single-page app.

Request validation happens before the route function runs. This is the body for starting an investigation:

# backend/moks/api/investigations.py
class CreateIn(BaseModel):
    question: str = Field(min_length=3, max_length=2000)
    # Both optional: MOKS resolves them from the question (see resolve_scope).
    environment: str | None = Field(default=None, pattern=r"^[a-z0-9][a-z0-9-]{0,99}$")
    namespace: str | None = Field(default=None, pattern=r"^(\*|[a-z0-9]([-a-z0-9]{0,61}[a-z0-9])?)$")
    minutes: int | None = Field(default=None, ge=5, le=10080)

A signed-in request with a one-character question or a namespace such as BAD! gets a 422 listing both fields, and the route never runs. An unsigned request gets a 401 instead, because the auth dependency raises first. The patterns also limit what can reach a Kubernetes or Loki query later.

Responses get far less attention. Only one route in the portal declares a response_model; the rest return dictionaries annotated dict[str, Any]. So the published schema says little about response shapes, and nothing filters a field a handler adds by accident. The serializers are hand-written instead.

Authentication and authorization as dependencies

Authentication is a FastAPI dependency, so a route can't reach its logic without it:

# backend/moks/auth/deps.py
async def get_auth(request: Request, db: AsyncSession = Depends(get_db, scope="function")) -> AuthContext:
    ctx = await load(db, request.cookies.get(COOKIE))
    if ctx is None:
        raise HTTPException(401, "Not signed in")
    s = get_settings()
    limiter.check(f"api:{ctx.session.id}", s.rate_api_per_minute)
    if request.method in UNSAFE:
        if not _origin_ok(request) or not csrf_ok(ctx.session, request.headers.get("x-csrf-token")):

In one place this covers the signed cookie session, a per-session rate limit, and an Origin and CSRF check on unsafe methods. Permission checks are separate and take an environment and namespace, because "may investigate" is meaningless without saying where. A denial writes an audit record.

Note scope="function". A yield dependency normally runs its exit code after the response is sent. My session dependency commits on exit, so with the default a client could read before the write landed. A docstring in the test suite records that this was observed live: someone could sign in and get a 401 immediately afterwards. The fix is one argument, and a test now fails if any code writes Depends(get_db) without it.

Async, blocking work and streaming

MOKS is async from top to bottom: SQLAlchemy over asyncpg, kubernetes_asyncio, HTTP clients. The exception is boto3, which blocks, so those calls go through asyncio.to_thread. FastAPI runs a plain def route in a thread pool but trusts an async def route not to block, so inside async def that job is mine.

Investigations take too long for one request, so the API doesn't run them. It writes a queued row and returns 201. A separate worker claims the row with FOR UPDATE SKIP LOCKED, and the browser follows progress over Server-Sent Events:

# backend/moks/api/investigations.py
    await db.commit()  # release the pooled connection; the stream uses short sessions

That line matters because the pool defaults to five connections, and a stream that held one for its whole life would exhaust it. The generator opens a short session per poll, runs for at most ten minutes, resumes from Last-Event-ID, and every five seconds rechecks the session and roles. If access has been revoked, the stream ends with a revoked event.

moks-terminal uses FastAPI's WebSocket support and checks the Origin header before accepting the upgrade. It shares the codebase but runs as its own process with its own routes and no OpenAPI endpoints. It is off by default in the chart.

RoboShop gateway: FastAPI as a transport shell

The gateway looks nothing like this. asgi_app.py registers four named routes and one catch-all, all pointing at the same function:

# services/ai-gateway/asgi_app.py
    app.add_api_route('/api/admin/llm/status', dispatch, methods=['GET'])
    app.add_api_route('/health/live', dispatch, methods=['GET'], response_model=HealthResponse)
    app.add_api_route('/health/ready', dispatch, methods=['GET'])
    app.add_api_route('/{path:path}', dispatch, methods=['GET','POST','PUT','PATCH','DELETE','HEAD','OPTIONS'], include_in_schema=False)

dispatch reads the raw body, enforces size limits and hands off to the existing Gateway.handle router. So FastAPI provides the ASGI interface, the lifespan, CORS middleware and a small OpenAPI document. Almost every URL the store uses is invisible to it.

That has a consequence that is easy to miss. ChatRequest is strict and forbids extra fields, but FastAPI never applies it, because dispatch takes a raw Request. The model appears in OpenAPI through openapi_extra, and the real check runs later inside the domain code:

# services/ai-gateway/server.py
            authenticate_service(req.headers)
            user = require_role(req.user)
            self.limit(req, True)
            data = req.body()
            message = text(validate(ChatRequest, data).message)

The order is deliberate. A comment in asgi_app.py states the principle: schemas must not bypass or disclose authentication decisions. I sent {"message": 5} to a real instance with a mocked store. Without the service token it returned 403. With the token it returned a generic 400. A FastAPI 422 never appears. I get authorization before validation, and I give up automatic error detail and most of the OpenAPI document.

The blocking work is handled differently too. MongoDB, Redis and SDK calls are synchronous, so dispatch admits a request to one of eight slots and runs it on a worker thread:

# services/ai-gateway/asgi_app.py
                with anyio.CancelScope(shield=True):
                    result = await anyio.to_thread.run_sync(invoke, abandon_on_cancel=False)

The shield is about payments. If a browser disconnects mid-checkout, the work finishes and the durable state machine stays in charge. A ninth concurrent request gets 503 Gateway busy instead of queuing.

FastMCP in practice

The Model Context Protocol is a standard way for an AI application to discover and call capabilities it wasn't built with. A server publishes tools, resources and prompts, and any compatible client can use them.

Which FastMCP this is

The name is ambiguous, so here is the exact answer for mcp-server. I use the standalone fastmcp package from PyPI, pinned to 4.1.3:

# services/mcp-server/server.py
from fastmcp import FastMCP
from fastmcp.tools import Tool, ToolResult
from fastmcp.server.auth import AccessToken, TokenVerifier

This is not the class that used to live at mcp.server.fastmcp. FastMCP's high-level API was incorporated into the official MCP Python SDK in 2024, and the two then moved apart. In the SDK version I pin, 2.3.0, that module no longer works: importing it raises an error pointing at the renamed MCPServer class. The official SDK is still present as a dependency, because FastMCP 4 is built on SDK v2, and my code imports mcp_types directly for content and elicitation types.

Host, client and server

MCP has three roles, and in my code only one is present.

My manifests give this server no in-cluster client. No workload carries the roboshop.ai/mcp-client label that its NetworkPolicy admits, and the shopping assistant does not use MCP. mcp-server never decides anything, since no model runs on its side of the protocol. It validates and forwards.

How tools are registered

Most FastMCP tutorials put @mcp.tool on a function and let type hints generate the schema. I don't do that. Each tool is a plain ToolSpec entry in gateway_tools.py with a hand-written JSON Schema, and one Tool subclass runs all of them:

# services/mcp-server/gateway_tools.py
    ToolSpec(
        "update_cart", "Set cart quantity",
        "Set the quantity of one SKU in the signed-in shopper's own cart; 0 removes it. Repeating the same call "
        "has no further effect. Does not reserve stock, take payment or place an order. Fails with a conflict "
        "while a payment is pending or when stock is insufficient.",
        {"sku": SKU, "quantity": {"type": "integer", "minimum": 0, "maximum": 20}},
        ["sku", "quantity"], CART,
        lambda args: ("POST", "/api/cart", None, {"sku": args["sku"], "quantity": args["quantity"]}),
        cart_view,
        roles=CUSTOMER, write=True, destructive=True, idempotent=True,
    ),

The catalogue holds no protocol code. Each entry carries the description, both schemas, how to build the gateway request and how to shape the answer. The description says what the tool will not do, which is what a model would otherwise guess. The quantity is absolute, so a repeated call doesn't double an order. Registration is a loop:

# services/mcp-server/server.py
    for spec in TOOLS:
        if not spec.confirm or admin_writes:
            server.add_tool(GatewayTool(spec, app))

Fifteen tools exist. Eleven are registered by default, because the four admin writes stay off unless MCP_ENABLE_ADMIN_WRITES=true, and the deployment sets it to false.

What FastMCP gives me here is everything around the catalogue: the protocol, the HTTP app, the TokenVerifier interface, per-tool auth checks, custom_route for health endpoints, the test client and telemetry hooks. I wrote the tool semantics. It wrote the plumbing.

Discovery, validation and errors

Over HTTP, each tool's auth callback filters the list by role. I connected the real server and client to a synthetic gateway and got 8 tools as a customer, 9 as an admin and 11 as a super admin. Calling a tool you can't see returns Unknown tool: 'admin_list_users', so its existence isn't confirmed.

Calling update_cart with quantity: "2" produced this, with the correlation ID shortened:

{"error": {"code": "invalid_argument", "message": "The arguments do not match this tool's input schema.", "retryable": false, "correlationId": "0f79..."}}

The gateway log for that call showed only the session check and no POST /api/cart. The message comes from the service's own validator, which refuses unknown fields, type coercion and control characters.

Failures are normal results with isError: true, as the specification recommends for tool execution errors, so a model can read them and react. retryable is never true for a write. That's an explicit signal not to repeat a change, and it matters more than the error text.

Output is validated too. Each result goes through a projection and a compiled JSON Schema check, and a malformed upstream response becomes invalid_response instead of reaching the model.

Transports and lifecycle

MCP_TRANSPORT picks one of two transports.

The HTTP options are short:

# services/mcp-server/server.py
    return {"path": "/mcp", "stateless_http": True, "json_response": True, "host_origin_protection": True,
            "allowed_hosts": hosts, "allowed_origins": origins}

With stateless_http, each request stands alone, so there is no session state to lose or leak between users. My client negotiated protocol revision 2026-07-28, which has no protocol-level session anyway. The cost is one GET /api/auth/me per request, deliberately uncached, so a revoked session fails on its next call. The only long-lived object is the pooled HTTP client, created lazily inside the event loop and closed by the server lifespan.

This server implements tools only. There are no resources, prompts or sampling, and elicitation exists only for the optional admin writes.

Two walkthroughs

RoboShop AI

RoboShop AI is a microservices store with an AI assistant. The gateway, MCP server and assistant are parts of one system, not separate projects.

One request. A user asks their AI client to add two units of a SKU to the cart. The client sends tools/call update_cart. HTTPTracing bounds the body at 256 KiB, FastMCP checks Host and Origin, and ShopSessionVerifier validates the bearer token against the gateway. GatewayTool.run checks the schema and per-user admission, then GatewayClient builds a new POST /api/cart. Only the accept, cookie: rs_auth=…, origin and trace headers leave the server. The gateway treats it like any browser call: session, Origin, ownership, stock and any pending payment. The result is shaped and validated on the way back. The write is never retried and is shielded from cancellation, so a timeout returns retryable: false.

Why it fits. The MCP server holds no business data and no privileged credential. It can only do what the signed-in user could already do.

Trade-offs. Each tool call costs two gateway requests. update_cart has no confirmation step, so a model can change a cart unless the host chooses to ask. The tool's reach is small, though: quantities in one cart, with no charge or reservation. And the gateway must run as a single process, because its JSON order ledger allows one writer.

MOKS

MOKS is an SRE investigation portal: an engineer asks what went wrong, and it gathers evidence from live systems and reports with citations.

One request. Figure 2 shows it. POST /moks/api/investigations passes CreateIn, get_auth, a rate limit and a scope check. The row, the first event and an audit record commit before the 201. A worker claims the row and runs the LangGraph investigation, and every tool call passes through ToolExecutor. The browser watches the event stream.

Why it fits. Splitting the API from the worker keeps slow investigations out of the request path, and PostgreSQL is the only queue, so there is one less system to run.

Why no MCP. MOKS has typed tools with Pydantic argument models, but they run in-process. The ToolExecutor owns authorization, audit, redaction and truncation, and the investigation code decides what to call. There is no third-party client to discover anything, so MCP would add a protocol layer and solve no problem I have.

Trade-offs. Progress is polled from PostgreSQL every half second per open stream, which is simple and resumable but not free. Rate limits are per replica. My docs/LIMITATIONS.md still lists streaming through the ingress as unchecked, and it says the terminal image has never run on a cluster, so read those parts as code rather than field results.

How they work together

FastMCP can be mounted inside a FastAPI app with http_app(), sharing one process if the lifespan is passed through. I don't do this. fastapi isn't even installed in the mcp-server environment. The two are separate images, pods and ASGI applications, and GatewayClient is an ordinary HTTP client.

Separation fits this code for concrete reasons. The gateway holds database and payment credentials, and the MCP server holds none, so a compromise there reaches only what a forwarded session allows. The MCP server has its own async HTTP client, while the gateway runs blocking work under a slot limit. The cost is another deployment, and a network hop for every tool call.

When a direct API call is enough

The RoboShop assistant calls the catalogue and gateway directly. The MOKS worker calls its tools in-process. Neither needs discovery, so neither uses MCP. Only mcp-server faces clients I can't predict, and that is the case the protocol exists for.

Discovery is not authorization

It's easy to treat a filtered tool list as an access control. It isn't one.

Over stdio, my server lists all eleven tools to everyone. A customer session sees admin_list_users and is refused by the gateway when it calls it. Over HTTP the list is filtered, but the docstring in gateway_tools.py is direct: "Roles only filter discovery and fail fast". The session is checked again at the gateway on the real call.

Tool annotations such as readOnlyHint aren't controls either. The specification says clients must treat them as untrusted unless they come from a trusted server. What protects the data is that the authorization decision sits in Gateway.handle, behind every path.

Deployment and operations

All four services run under Uvicorn. The two RoboShop images use python:3.14-slim, and MOKS uses python:3.12-slim. Workloads are defined as manifests or a Helm chart and delivered by Argo CD.

Logs are structured JSON on all of them. Both projects propagate trace context. mcp-server passes span attributes through an allowlist, because FastMCP may attach tool arguments. On EKS the RoboShop config currently sets OTEL_SDK_DISABLED=true, so that tracing exists in code and is switched off there.

mcp-server runs its unit tests during the image build, so a failing contract test stops the image. I ran its 20 tests with the pinned packages on Python 3.11 and all passed. The gateway's FastAPI, chat history and user-service tests passed too, all 28. Six admin-console tests errored in their fixture with mongomock 4.3.0, which the service doesn't pin, so those remain unverified. The MOKS backend gave 835 passed, 3 failed and 4 skipped; the failures are Helm render tests for terminal defaults.

Read tools, write tools and least privilege

In place today:

• Read tools can retry once on a transient failure, inside one deadline. Write tools never retry.
• update_cart is the only write a customer can reach.
• Admin writes aren't registered by default. When enabled, a model's boolean isn't accepted as approval. The first call returns an elicitation with HMAC-signed state bound to the caller, the tool, a hash of the arguments, a two-minute expiry and a one-time nonce.
• No tool takes a URL, SQL, a shell command or a payment action.
• MOKS routes ToolExecutor denials into the audit log and checks namespace scope on each call.

That signed confirmation has a stated limit. Its docstring says it proves the confirmation came through the authenticated client for those arguments, and that it can't prove a person clicked.

Lessons and improvements

Keep the authorization decision below the protocol. This is the most useful pattern in the code. A new MCP tool can't grant anything the gateway refuses.

Treat validation as layers. A strict tool schema protects the gateway from a model's mistakes, and the gateway's own checks protect everything from the MCP server's mistakes.

A framework only works where you let it. In MOKS, FastAPI validates every body and generates a useful schema. In the gateway, one catch-all route bypasses most of what the framework offers. The gateway's approach kept existing handlers and their authorization order intact, but the OpenAPI document covers only four paths.

These are recommendations, not current behaviour:

1. Turn off the gateway's default docs. /docs, /redoc and /openapi.json respond at the application. NGINX only proxies /api/, so they aren't routed through the storefront, but the MOKS pattern of disabling them by default is safer.
2. Add OAuth before any public MCP endpoint. A bearer that is an ordinary shop session has the user's full authority for up to eight hours. That doesn't meet the MCP authorization profile, and the README already forbids exposing /mcp.
3. Handle hostile tool output. Product descriptions reach the client's model unmodified, while the specification says servers must sanitize tool outputs. Structure checks don't address intent.
4. Add an inbound mesh policy for mcp-server before enforcing allow-nothing, or a future in-cluster client will be blocked.
5. Add response models to MOKS for routes that return user or security data, so a stray field can't leak.
6. Share rate limits across replicas if MOKS scales past a small number of API pods.

If I had to give one piece of advice from this code, it would be to decide where authorization lives before choosing a protocol. FastAPI and FastMCP both make it easy to expose things. Neither decides who should be allowed to use them.