Building and securing MCP servers with FastMCP

Most MCP tutorials stop at a server that adds two numbers over stdio. That’s the easy 10%. The hard part, and the part nobody writes up, is everything you hit the moment you want to put an MCP in front of real company data: who’s allowed to call what, how you authenticate them, what you log, and how you stop the model drowning in eighty tools it doesn’t need.

I’ve spent a big chunk of the last few months living in this, including a run of conversations with Australian businesses and a few US fintech companies trying to wire MCPs into real systems. The question that comes up every time is less about whether you can build a tool and more about whether you can safely let one reach production data, and the fintechs especially won’t let you hand-wave it.

So this is the build-along I wish I’d had. I’ll start from a local toy and grow it into something you could actually deploy, using FastMCP 3, the Python framework that handles most of the protocol plumbing for you. It covers auth, per-group access, logging, multi-domain design, the deterministic-tool-versus-agent question, getting files to people without stuffing them through the model, and testing the lot without a real identity provider or an AWS account.

A server in ten lines #

Install it with pip install fastmcp, then a tool is just 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 reads the type hints and docstring and turns them into the schema the model sees, so the signature you’d write anyway is the contract. That’s the whole appeal: you write normal typed Python and the framework deals with the protocol.

Local versus remote #

That mcp.run() with no arguments speaks over stdio, which means the server runs as a subprocess of whatever launched it, on the same machine, as that user. It’s perfect for a local tool wired into Claude Desktop or an editor, and it inherits the user’s own machine and credentials, so there’s nothing to authenticate.

The moment you want more than one person to use it, you go remote:

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

Now it’s a web service, and everything that’s true of any web service in front of company data is suddenly true of your MCP. It’s reachable, so it needs authentication. It serves many people, so it needs to know who each one is. I run these on Fargate behind an ALB, the same as any other internal service, and the rule I’d hold onto is simple: a local stdio server is a convenience, a remote HTTP server is production infrastructure and gets treated like it.

You don’t need two codebases for the two postures, though. In the repo it’s a single env var: main() runs mcp.run() over stdio by default, and switches to HTTP when ACME_MCP_REMOTE is set. Same server, two ways to run it.

Putting auth on it #

A remote MCP with no auth is an open door to whatever it can reach, so this is the bit to get right first. FastMCP 3 takes an auth provider on the server and verifies every request before your tools ever run. For anything real you want a JWT verifier pointed 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 a quick local spike there’s a StaticTokenVerifier you can hand a dict of tokens to, but keep that for development. There are also full OAuth providers if you want the server to run the login dance 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 identity."""
    token = get_access_token()
    return {
        "user": token.claims.get("sub"),
        "groups": token.claims.get("groups", []),
    }

That claims dict is the user’s verified token, and it’s the foundation everything else builds on, because it tells you not just that someone is allowed in, but who they are and what they’re entitled to. And the public tag I put on whoami matters once you start filtering tools by group, which is the next section.

Hiding tools by user group #

Here’s where a company MCP earns its keep. You don’t want the support team seeing finance tools, and you definitely don’t want a refund tool showing up for someone who can only read orders. The trick is to tag tools by the domain they belong to, then filter what each caller sees based on their groups.

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 resolve the current caller’s groups into the tags they may use:

from fastmcp.server.dependencies import get_access_token

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 allowed_tags() -> set[str]:
    token = get_access_token()
    if token is None:
        return set()  # unauthenticated: see nothing
    groups = token.claims.get("groups", [])
    tags = set().union(*(GROUP_TAGS.get(g, set()) for g in groups), set())
    return tags | PUBLIC_TAGS

Two things there are the hardening, not the first draft. admin maps to {ALL_TAGS}, a "*" wildcard, rather than the explicit {"orders", "billing", "admin"} I wrote first. The explicit list is a trap: the moment you compose in a later domain like the proxied analytics service further down, admin silently stops seeing it because nobody remembered to add analytics to the list. The wildcard keeps admin a true superset as domains get mounted, no edits required.

And PUBLIC_TAGS exists because whoami has no business-domain tag, so under a filter that only passes the tags a group is cleared for, it would disappear for everyone, admin included. Identity and health tools need their own “any authenticated caller” tier, which is what the public tag on whoami buys. The default for an unauthenticated caller is still nothing.

Which shakes out to this, and it’s worth seeing the whole matrix at once:

The empty bottom row is the default, and the admin row is a wildcard rather than five ticks I have to keep in sync.

Now enforce it. In FastMCP 3 this is middleware, which runs per request so it can see the authenticated caller and decide on their groups. My first cut was a Transform, and I’ve seen that sketch copied around, but middleware is the supported per-request mechanism, so this is the version that actually runs. The filter enforces in two places:

from collections.abc import Sequence
from fastmcp.exceptions import ToolError
from fastmcp.server.middleware import Middleware, MiddlewareContext

def cleared_for(tool_tags, allowed) -> bool:
    """Whether a caller cleared for `allowed` may use a tool tagged `tool_tags`."""
    if ALL_TAGS in allowed:
        return True
    return bool(set(tool_tags) & allowed)

class GroupTagFilter(Middleware):
    """Show and run only the tools a caller's groups allow."""

    async def on_list_tools(self, context: MiddlewareContext, call_next) -> Sequence:
        tools = await call_next(context)
        tags = allowed_tags()
        return [t for t in tools if cleared_for(t.tags, tags)]

    async def on_call_tool(self, context: MiddlewareContext, call_next):
        name = context.message.name
        tool = await context.fastmcp_context.fastmcp.get_tool(name)
        if not (tool and cleared_for(tool.tags, allowed_tags())):
            raise ToolError(f"Unknown tool: {name}")
        return await call_next(context)

mcp.add_middleware(GroupTagFilter())

The two layers do different jobs. on_list_tools hides the tools a caller shouldn’t see, so they never clutter the model’s context. on_call_tool is the one that actually protects you: it blocks a call to a disallowed tool even if the model guesses the name, so a hidden tool is genuinely uncallable rather than merely invisible. Filter only the list and a guessed name still runs; filtering the call is what shuts the door.

The other detail worth copying is the error. A blocked call raises the same ToolError("Unknown tool: ...") a nonexistent tool would, so a caller can’t tell “you’re not cleared for this” apart from “this doesn’t exist”, and you don’t leak the existence of tools they aren’t cleared for.

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

flowchart LR
    A[AI agent] -->|request + JWT| B{Verify token}
    B -->|invalid| X[401 rejected]
    B -->|valid| C[Audit-log middleware]
    C --> D{Group filter}
    D -->|allowed for caller| E[Tool runs]
    D -->|not allowed| Y[Tool not found]
    E --> F[(DynamoDB / internal API)]

Agents behind tools, or just tools #

A question that comes up fast: should a tool do a deterministic thing, or should it front an AI agent?

A deterministic tool is the boring, correct default. It does one well-defined thing and returns data, like a lookup against DynamoDB 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"]

You can read it, test it, and reason about its blast radius, and it costs a database call.

Boring and correct still means validating what comes in, and the tool that moves money validates before it writes. issue_refund rejects a blank order id and any non-positive or non-finite amount, and it explicitly rejects booleans, because bool is a subclass of int in Python and True would otherwise pass as 1.0:

@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

Sometimes you genuinely want reasoning inside the tool, and then you can put an agent behind it. Make that inner agent an injected interface rather than a hard-wired call, so the tool never changes whether it’s backed by a real model, a stub, or a test fake:

import typing

@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."""
    await ctx.info(f"drafting refund email for {order_id}")
    prompt = f"Write a brief, on-brand refund-confirmation email for order {order_id}."
    return await _agent.run(prompt)

The ctx parameter is FastMCP’s per-call context, injected for you, and await ctx.info(...) records a per-call note against the call. Putting an agent behind a tool is powerful, but you’ve now got a model calling a tool that calls another model, which costs more and runs slower, and gets a lot harder to test and predict.

My rule is to default to deterministic tools and only put an agent behind one when the task genuinely needs open-ended reasoning you can’t express as code. When you do, keep that inner agent on a tight leash, with its own narrow tools and credentials and a clear boundary, so a confused outer model can’t talk it into something daft. The injection seam is part of that leash: the boundary is visible in the code, you swap a Claude-backed agent in at startup with set_agent, and a test swaps in a fake that makes no network call at all.

Multi-domain design and lazy loading #

Models get worse as you pile on tools. Give one fifteen and it picks well; give it eighty across five unrelated domains and it starts reaching for the wrong one. So if your MCP covers billing and support and analytics and everything else, the goal is to keep the surface any one caller sees small and relevant.

The group filter above already does a lot of this, since someone in support never loads the finance tools. You can lean on it further by composing servers per domain rather than writing one giant file. FastMCP lets you mount a sub-server’s tools 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 mount the domains without a namespace prefix, so the tools stay order_status, not orders_order_status. These are domains of one product, not third-party servers you’re disambiguating, so the clean names are the right call. The one thing to be careful of is the proxy: wire create_proxy(...) at import time and your server can’t boot unless the analytics service is reachable, which is a bad dependency to bake in, so it’s mounted only when its URL is set.

Which gives you this:

flowchart LR
    C[Client] --> M[Main MCP server]
    M --> O[Orders tools]
    M --> B[Billing tools]
    M -.->|proxy| A[Analytics MCP · separate service]
    O --> DB[(Orders DB)]
    A --> DW[(Analytics warehouse)]

The proxy is the closest thing to lazy loading: the analytics domain runs as its own service, owned by its own team, and the main server forwards to it rather than holding it all in process. That keeps each domain independently deployable and stops a single server becoming a monolith that everyone is scared to touch. If two domains have nothing to do with each other and serve different teams, the honest answer is often two separate servers behind their own auth, not one server trying to be everything.

Logging and the audit trail #

Once an MCP is brokering authenticated access to company data, “who called what, and when” stops being nice-to-have. For per-call notes inside a tool there’s the context object, with await ctx.info(...) as you saw above. For the cross-cutting audit trail you want middleware, which wraps every tool call without you touching each tool:

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())

Two fields there earn their place. Logging the caller’s groups alongside their sub is the difference between “who did it” and “what were they cleared as”, and the second is what tells you after the fact whether an access decision was right. And catching the exception to record its type, then re-raising, means denied and failed calls land in the trail too, not just the ones that succeeded. Ordering matters for the same reason: add AuditLog before GroupTagFilter so audit wraps the outermost call and a blocked attempt still gets logged, rather than being swallowed by the filter before audit ever sees it.

Ship those logs to CloudWatch and you’ve got the audit trail you’ll be very glad of the first time someone asks who pulled a particular customer’s record. It’s also where you’d hang rate limiting and anomaly checks later, since middleware sees every call in one place.

Getting files to people #

A tool that returns a 40MB PDF as text is a tool that blows up the model’s context and costs a fortune. Don’t do it. When a tool produces a file, put the file in S3 and hand back a short-lived signed URL instead:

import boto3

s3 = boto3.client("s3")

@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 "/" in report_id or "\\" in report_id or ".." in report_id:
        raise ToolError("report_id must not contain path separators")

    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=300,  # link dies in five minutes
    )
    return {"download_url": url, "expires_in": 300}

The model never sees the bytes, just a link it can pass to the user, and the presigned URL carries its own short-lived, scoped permission to that one object. Keep the expiry tight and make sure the role doing the signing can only touch that export prefix, not the whole bucket. That prefix scoping is also why report_id gets validated first: it’s interpolated straight into the S3 key, so a crafted id with .. or a slash in it could walk the signed URL outside reports/ and defeat the very scoping you set up in IAM. Two lines of guard close that off.

The bytes go around the model, not through it:

flowchart LR
    Ag[AI agent] -->|call export_report| T[Tool]
    T -->|upload file| S3[(S3)]
    T -.->|signed URL only| Ag
    Ag -->|passes link| U[User]
    U ==>|downloads direct| S3

This is also where a skill earns its place alongside the MCP. Bundle a small skill that tells the agent how to behave when a tool returns a download_url: present it to the user as a link, mention the expiry, and don’t try to read the file into context. Two rules are worth spelling out. A signed URL is a bearer credential, so the agent should never log, echo, or forward it beyond the person who asked. And once a link has likely expired it should call the tool again for a fresh one rather than reusing the old URL. The MCP delivers the file safely; the skill supplies the direction for what to do with it.

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

The fintechs won’t hand-wave testing any more than they’ll hand-wave auth, and neither should this post. The nice thing is that every piece above is testable with no network: no real identity provider, no live JWTs, no AWS account.

FastMCP ships an in-memory Client that drives a server object directly, so a test calls tools exactly as a model would but in-process. You simulate an authenticated caller by setting the auth context var to an AccessToken carrying whatever claims you want, which is the documented way to test auth without minting a real 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 one test proves the layer that actually protects you: naming a hidden tool directly still fails. The rest of the access matrix is the same shape. It’s worth asserting the default-deny corners on their own, that an unauthenticated caller sees nothing and an unknown group sees only the public tools, because “it defaults to closed” is a much stronger claim with a test behind it than with a comment.

The S3 path uses moto to fake the bucket, so export_report really uploads and really signs a URL, all in memory, and the test can assert both that the object landed and that the returned dict carries no raw bytes. The agent-backed tool takes the injection seam from earlier: set_agent swaps in a fake that records its prompt and returns a marker, so you assert the tool forwarded the order id and returned the agent’s text without ever calling a real model. The whole example runs green on a laptop with nothing configured, which is exactly what you want the day someone asks you to prove the access rules do what you say.

Skills or a server #

Having built all that, the honest question is whether you needed a server at all, or just a skill, and for a lot of jobs the answer is a skill, so I reach for one first when it’s direction or local glue. It stops being enough the moment the data is shared and access to it has to be governed, because a skill runs in the user’s own context with no server in the middle to authenticate a caller, hide tools by group, or keep an audit trail. That’s the whole reason this server exists. The file-delivery example above is the two working together, a skill steering an agent that’s driving a properly secured server, and I wrote up where I draw the line between them in a separate post.

Where this leaves you #

The toy server is ten lines and you’ll have it running in a minute. Everything past that is the distance between a demo and something you’d let near customer data: authenticating every caller, filtering tools so the call itself enforces access, logging who did what, keeping agents on a leash, and handing over files by reference. None of it is hard with FastMCP doing the plumbing, but it’s the part that lets you defend the thing in a review.

I’ve put the whole thing together in acme-mcp, wired up and runnable, so you can see how the pieces fit instead of stitching the snippets together yourself.

I specced and prompted a good chunk of this server up in ‘Kiro’, AWS’s spec-driven IDE, then hardened it by hand, which felt about right for a post on letting models reach into real systems. The first draft was mostly fine, and the interesting parts are where it wasn’t: the admin group with an explicit tag list that would quietly stop seeing new domains, the refund tool that took an amount without checking it, the report id that walked straight into an S3 key. The AI is great at the first draft, and the judgement about what to lock down is still the job.