Building and securing MCP servers with FastMCP

Most MCP tutorials stop at a few tools running on your own machine with no authentication. That is enough to learn the protocol because you are the only person who can call them. Company data already has access controls around it, so a tool that answers anyone who asks bypasses those controls.

Over the last few months I’ve helped Australian businesses and US fintechs work through this. The shape I keep coming back to starts with a local FastMCP 3 server and grows it into something you could deploy. It checks group access and records calls. For file delivery, it returns a link instead of pushing the file through model context. The example still runs without an identity provider or AWS account.

A server in ten lines #

Install it with pip install fastmcp. A tool is a typed Python function with a decorator:

from fastmcp import FastMCP

mcp = FastMCP("acme-tools")

@mcp.tool
def order_status(order_id: str) -> dict:
    """Look up the current status of an order."""
    return {"order_id": order_id, "status": "shipped"}

if __name__ == "__main__":
    mcp.run()  # stdio by default

FastMCP uses the type hints and docstring to build the schema the model sees.

Local versus remote #

An MCP server can run locally beside the agent or remotely as a shared service. That choice decides whose credentials it uses and where access is enforced.

The ten-line example above runs locally. mcp.run() starts the server as a subprocess of the MCP client, and they communicate over stdin and stdout. MCP calls this the stdio transport. There is nothing else to deploy or expose on a network, and the server inherits your machine’s credentials. For personal tooling and development, this is the right default.

Each user installs and maintains their own copy. It runs with the credentials already on their machine, so the MCP layer has no central place to enforce what they can do with the data behind it.

The same server can run over HTTP:

mcp.run(transport="http", host="0.0.0.0", port=8000)

Over HTTP, the MCP server becomes a web service in front of company data. One deployment serves everyone, so users don’t install or update it themselves. The server authenticates each caller against your identity provider, and shared data gets one enforcement point with an audit log. It is now an internal service like any other. I run these on Fargate behind an ALB and secure them accordingly.

The repo uses one codebase: stdio by default, and HTTP when ACME_MCP_REMOTE is set.

Putting auth on it #

A remote MCP server with no authentication lets anyone who can reach it use its tools. FastMCP 3 takes an auth provider and verifies each request before a tool runs. For production, point a JWT verifier at your identity provider’s public keys:

from fastmcp import FastMCP
from fastmcp.server.auth.providers.jwt import JWTVerifier

auth = JWTVerifier(
    jwks_uri="https://auth.acme.internal/.well-known/jwks.json",
    issuer="https://auth.acme.internal",
    audience="acme-mcp",
)

mcp = FastMCP("acme-tools", auth=auth)

For local development, StaticTokenVerifier can use a dictionary of tokens, but I keep it out of production. FastMCP also has OAuth providers if you want the server to run the login flow itself.

Once a request is authenticated, any tool can ask who’s calling:

from fastmcp.server.dependencies import get_access_token

@mcp.tool(tags={"public"})
def whoami() -> dict:
    """Return the caller's verified identity and groups."""
    token = get_access_token()
    claims = token.claims if token else {}
    return {
        "user": claims.get("sub"),
        "groups": claims.get("groups", []),
    }

Those claims come from the user’s verified token. They tell the server who is calling and what they may use. The public tag keeps whoami available after the server starts filtering tools by group.

Hiding tools by user group #

The server enforces group access. Support may need billing lookups, but a read-only caller should never receive issue_refund in tools/list. I tag each tool by domain and map the caller’s groups to the tags they may use.

Tag the tools as you define them:

@mcp.tool(tags={"orders"})
def order_status(order_id: str) -> dict:
    """Read-only order lookup."""
    ...

@mcp.tool(tags={"billing"})
def get_invoice(invoice_id: str) -> dict:
    ...

@mcp.tool(tags={"admin"})
def issue_refund(order_id: str, amount: float) -> dict:
    ...

Then map each org group to the tags it’s cleared for, and turn a token’s groups claim into the tags it permits:

ALL_TAGS = "*"  # wildcard: every tag, including ones from domains added later

# which tool tags each org group is allowed to see and call
GROUP_TAGS = {
    "support": {"orders", "billing", "support", "reports"},
    "finance": {"billing", "reports"},
    "admin": {ALL_TAGS},
}

# tags any authenticated caller gets, whatever their group
PUBLIC_TAGS = {"public"}

def tags_for_groups(groups) -> set[str]:
    """Resolve a groups claim into the set of tool tags it permits."""
    tags = set().union(*(GROUP_TAGS.get(g, set()) for g in groups))
    return tags | PUBLIC_TAGS

I originally listed every admin tag explicitly. When I mounted analytics, admin lost access because I had not added the new tag. Mapping admin to * removes that maintenance step.

PUBLIC_TAGS keeps identity and health tools visible to any authenticated caller. Without it, whoami has no business-domain tag and disappears under the group filter. An unauthenticated caller still sees nothing.

The resulting access matrix looks like this:

The empty bottom row is the default. No token means no tags, so an unauthenticated caller sees nothing. The admin row comes from the * wildcard in GROUP_TAGS. When I mount a new domain, admin picks up its tools without another change to the mapping.

FastMCP 3 accepts a callable authorization check. It receives the caller’s token and the component being accessed, then returns True or False:

from fastmcp.server.auth import AuthContext
from fastmcp.server.middleware import AuthMiddleware

def group_access(ctx: AuthContext) -> bool:
    """Allow a caller whose groups clear them for the component's tags."""
    if ctx.token is None:
        return False  # unauthenticated: see nothing
    allowed = tags_for_groups(ctx.token.claims.get("groups", []))
    if ALL_TAGS in allowed:
        return True
    component_tags = set(ctx.component.tags)
    if skill_info := getattr(ctx.component, "skill_info", None):
        component_tags.update(skill_info.frontmatter.get("tags", []))
    return bool(component_tags & allowed)

mcp.add_middleware(AuthMiddleware(auth=group_access))

AuthMiddleware uses the check when listing and calling components. Denied tools stay out of tools/list, and naming one directly still fails. The skill_info fallback reads tags from skill frontmatter because FastMCP 3.4 keeps them as provider metadata rather than normal component tags.

Denied tools do not disappear completely. Calling one directly returns insufficient permissions, while a missing tool returns not found. A caller can use that difference to confirm a tool name. If that matters for your threat model, normalise both errors to Unknown tool in another middleware. The built-in denial is enough for this example.

Put together, a request passes several checks before it reaches a backend:

flowchart TD
    req["AI agent: request + JWT"] --> verify{"verify token"}
    verify -->|invalid| rej["401 rejected"]
    verify -->|valid| audit["audit-log middleware"]
    audit --> auth{"auth middleware"}
    auth -->|allowed| tool["tool runs"] --> backend[("DynamoDB / internal API")]
    auth -->|"not allowed"| denied["not authorized"]

Agents behind tools, or just tools #

I default to a deterministic tool with one defined job, such as a DynamoDB lookup or a call to an internal API:

@mcp.tool(tags={"orders"})
def order_status(order_id: str) -> dict:
    """Return the current status for an order (read-only)."""
    return orders_table.get_item(Key={"id": order_id})["Item"]

The test calls it and checks the result. Its blast radius is visible in the code.

The tool that moves money validates before it writes. issue_refund rejects a blank order id and requires a finite amount above zero. It rejects booleans explicitly because bool is a subclass of int in Python:

@mcp.tool(tags={"admin"})
def issue_refund(order_id: str, amount: float) -> dict:
    """Issue a refund against an order (privileged write)."""
    if not order_id or not order_id.strip():
        raise ToolError("order_id is required")
    if isinstance(amount, bool) or not isinstance(amount, (int, float)):
        raise ToolError("amount must be a number")
    if not math.isfinite(amount) or amount <= 0:
        raise ToolError("amount must be a positive, finite number")
    ...  # record the refund

Some tasks need open-ended reasoning. I inject the inner agent so production can provide a model-backed implementation while tests use a fake:

import re
import typing

_SAFE_ORDER_ID = re.compile(r"[A-Za-z0-9._-]+")
MAX_ORDER_ID_LEN = 64

@typing.runtime_checkable
class Agent(typing.Protocol):
    async def run(self, prompt: str) -> str: ...

_agent: Agent = StubSupportAgent()  # deterministic default; no LLM to boot

def set_agent(agent: Agent) -> None:
    global _agent
    _agent = agent

@mcp.tool(tags={"support"})
async def draft_refund_email(order_id: str, ctx: Context) -> str:
    """Draft a customer email about a refund."""
    if not order_id or not order_id.strip():
        raise ToolError("order_id is required")
    if len(order_id) > MAX_ORDER_ID_LEN or not _SAFE_ORDER_ID.fullmatch(order_id):
        raise ToolError("order_id may only contain letters, digits, '.', '-', and '_'")
    await ctx.info(f"drafting refund email for {order_id}")
    prompt = (
        "Write a brief, friendly refund-confirmation email to the customer for "
        f"order {order_id}. Keep it warm and on-brand; do not promise a refund "
        "amount or date."
    )
    return await _agent.run(prompt)

FastMCP injects the per-call ctx, and await ctx.info(...) records a note against the call. The order id goes into the inner model’s prompt, so the allowlist rejects spaces and control characters first.

An agent-backed tool costs more and takes longer than a deterministic call. I use one when the task needs it, with scoped credentials and a narrow tool set. The injected Agent keeps tests off the network.

Multi-domain design and lazy loading #

Every tool has a cost before the conversation starts because clients usually send the catalog of schemas to the model up front. With forty tools across five domains, that uses context for tools the caller will never need and gives the model more wrong options to choose from. I keep each caller’s catalog to the tools that matter to them.

The access filter does most of that work because someone in support never loads the finance tools. FastMCP also lets me keep each domain as its own server and mount them into a parent:

import os
from fastmcp import FastMCP
from fastmcp.server import create_proxy

main = FastMCP("acme")
for sub in (orders_server, billing_server, support_server, reports_server):
    main.mount(sub)  # in-process domains, no namespace prefix

# a heavy or separately-owned domain, proxied from its own service.
# only mounted when it's configured, so the server still boots offline
if url := os.environ.get("ACME_MCP_ANALYTICS_URL"):
    main.mount(create_proxy(url))

I leave off namespace prefixes because these domains belong to one product, so the tool stays order_status rather than orders_order_status. Analytics is only mounted when ACME_MCP_ANALYTICS_URL is set. An unreachable proxy cannot stop the main server from starting.

The resulting shape is:

flowchart LR
    client["client"] --> main["main MCP"]
    main --> orders["orders tools"] --> odb[("orders DB")]
    main --> billing["billing tools"]
    main -.->|proxy| analytics["analytics MCP, a separate service"] --> wh[("analytics warehouse")]

Analytics runs as its own service and the main server forwards calls to it. I split out a domain when another team owns it or it needs its own authentication boundary. The access filter handles what each caller sees. The tool-design post covers larger catalogs.

Logging and the audit trail #

In production it’s important to know who called each tool and when. ctx.info(...) records notes inside one tool. Middleware wraps every call with an audit record:

import time
import logging
from fastmcp.server.middleware import Middleware
from fastmcp.server.dependencies import get_access_token

log = logging.getLogger("mcp.audit")

class AuditLog(Middleware):
    async def on_call_tool(self, context, call_next):
        token = get_access_token()
        user = token.claims.get("sub") if token else "anon"
        groups = token.claims.get("groups", []) if token else []
        tool = context.message.name
        start = time.perf_counter()
        error = None
        try:
            return await call_next(context)
        except Exception as exc:
            error = type(exc).__name__
            raise
        finally:
            ms = round((time.perf_counter() - start) * 1000)
            log.info("tool_call", extra={"user": user, "groups": groups, "tool": tool, "ms": ms, "error": error})

mcp.add_middleware(AuditLog())

I log the token’s groups with its sub because the group claim explains why the caller received access. AuditLog sits before the access middleware, so it sees blocked attempts and records exception types before re-raising them.

I send those logs to CloudWatch so I can query who pulled a particular customer’s record. If I add rate limiting or anomaly checks later, middleware gives them the same view of each call.

Getting files to people #

Not every tool result belongs in the context window. A small lookup is fine as JSON. A 40MB PDF returned as text fills the model’s context and costs a fortune; an Athena query returning a hundred thousand rows has the same problem.

I choose the delivery path based on the size and sensitivity of the result:

Large results can be scoped before delivery too. The server can run a caller-scoped Athena query and write the result as a new S3 object. A short-lived signed URL works when bearer-link access fits the threat model; otherwise the export goes through the authenticated front door.

acme-mcp’s report export follows the large, non-sensitive path. The tool writes the file to S3 and hands back a short-lived signed URL instead of the bytes:

import re

import boto3

s3 = boto3.client("s3")
SAFE_REPORT_ID = re.compile(r"[A-Za-z0-9._-]+")
MAX_REPORT_ID_LEN = 128

@mcp.tool(tags={"reports"})
def export_report(report_id: str) -> dict:
    """Generate a report and return a short-lived download link."""
    if not report_id or not report_id.strip():
        raise ToolError("report_id is required")
    if len(report_id) > MAX_REPORT_ID_LEN:
        raise ToolError("report_id is too long")
    if not SAFE_REPORT_ID.fullmatch(report_id):
        raise ToolError("report_id may only contain letters, digits, '.', '-', and '_'")

    key = f"reports/{report_id}.pdf"
    # ... build the file and upload it to S3 under `key` ...
    url = s3.generate_presigned_url(
        "get_object",
        Params={"Bucket": "acme-mcp-exports", "Key": key},
        ExpiresIn=60,  # lowest expiry supported by this delivery path
    )
    return {"download_url": url, "expires_in": 60}

The model sees a link rather than a big JSON object.

This delivery path has a minimum expiry of 60 seconds, so that is what I use here. The URL passes through the agent’s context and logs, and anyone who reads either can use it until it expires. The signing role should only reach the export prefix.

S3 object keys are not filesystem paths, so .. does not traverse out of a prefix. The allowlist keeps the generated key predictable and rejects separators or control characters before they reach S3 and the logs.

The user downloads straight from S3 and the model only ever handles the link:

sequenceDiagram
    participant Agent as AI agent
    participant Tool as tool
    participant S3
    participant User as user
    Agent->>Tool: call export_report
    Tool->>S3: upload the file
    Tool-->>Agent: signed URL only
    Agent->>User: passes the link
    User->>S3: downloads direct

A short expiry limits exposure but does not identify the person holding the link. For sensitive files, I bind the download to an identity:

• For a person clicking the link, return your own authenticated endpoint. It checks the user’s SSO session and confirms they own the export before redirecting them to S3. Pinning the requester’s IP can work on a fixed network, but breaks down behind corporate NAT or on mobile.
• For application access, use S3 Access Grants with identity propagation to map corporate identities to S3 prefixes.

A small skill tells the agent to show download_url and mention the expiry. It also tells the agent to leave the file and URL out of its logs, then request a fresh link after expiry. The server controls access to the file, while the skill controls how the agent handles it. I wrote more about when I use each.

The skill can live in the server too. FastMCP’s SkillProvider publishes a skill directory as MCP resources, so the client can discover and read the instructions without a separate install:

from pathlib import Path

from fastmcp.server.providers.skills import SkillProvider

handle_downloads = Path(__file__).parent / "skills" / "handle-downloads"
mcp.add_provider(SkillProvider(handle_downloads))

In the repo, the skill carries the reports tag and appears at skill://handle-downloads/SKILL.md. The same middleware used for report tools checks the skill’s frontmatter tag, so callers with reports access can read it. The client fetches the skill separately when it wants the handling instructions.

Testing all of this without an IdP or an AWS account #

You can test this without network access or AWS credentials.

FastMCP’s in-memory Client drives the server object directly, so tests call tools through the same interface as a model. In unit tests I set the auth context directly to an AccessToken. To exercise token verification as well, use StaticTokenVerifier or mint a throwaway JWT.

import contextlib
from mcp.server.auth.middleware.auth_context import (
    AccessToken, AuthenticatedUser, auth_context_var,
)

@contextlib.contextmanager
def as_caller(groups):
    token = AccessToken(token="t", client_id="u", scopes=[],
                        claims={"sub": "u@acme.test", "groups": groups})
    reset = auth_context_var.set(AuthenticatedUser(token))
    try:
        yield
    finally:
        auth_context_var.reset(reset)

async def test_support_cannot_call_admin_tool(server):
    with as_caller(groups=["support"]):     # support is not cleared for admin
        async with Client(server) as client:
            with pytest.raises(Exception):
                await client.call_tool("issue_refund", {"order_id": "A1", "amount": 5})

That test proves that naming a hidden tool directly still fails. Separate tests cover the default-deny cases, including unauthenticated callers and unknown groups.

moto supplies the S3 bucket, so the test uploads an object and checks that the result contains only a URL. A fake agent records its prompt and returns a marker, which lets the full example run on a laptop with no external services configured.

The complete example #

The complete acme-mcp example is wired up and runnable, so you can see how the pieces fit without stitching the snippets together yourself.

I specced and prompted much of acme-mcp with OpenSpec and Pi, then hardened it by hand. Admin access was the first miss because its explicit tag list went stale as soon as I added a domain. The refund amount and report id also needed boundary checks. The first draft came together quickly, but I still had to find those missing access and validation checks.