Build an MCP Gateway in Python: Least-Privilege Tools per Agent, and What It Costs

Once you connect more than one MCP server to an agent, two problems show up. Every agent sees every tool, including the ones that write, reset or reach the network. And every tool definition is sent to the model on every turn, whether the agent needs it or not. Searches that reach our MCP explainer keep asking the same thing: how do you give each agent only the tools and access it should have?

A common answer is an MCP gateway: one MCP server that sits in front of the others, shows each agent role a filtered tool list, and checks every call before passing it on. Commercial and open-source gateways exist, but most write-ups are product lists. This post builds a small one in Python (about 150 lines, official MCP SDK 2.x), puts it in front of three official reference servers (git, fetch, time), and measures what it saves and what it costs.

Every number comes from the scripts on this page, run three times on an Apple M4 laptop (16 GB RAM, macOS 26.6.2, Python 3.12.14, mcp 2.3.0, reference servers 2026.11.4, stdio transport, everything local).

Short version:

• Filtering tools per role cut the definitions the model sees by 57–69%. All 18 tools cost 3,919 tokens (tiktoken o200k_base). A read-only reviewer role saw 7 tools and 1,216 tokens. A web researcher role saw 4 tools and 1,693.
• The extra hop added about 0.6 ms per call (median 0.62 → 1.19–1.20 ms for a trivial tool). For a 13–14 ms git status it was within the run-to-run noise.
• Startup is the real cost. The gateway starts its upstream servers in turn: 0.74–0.76 s for a role that needs two servers, 0.98–1.05 s for all three, against 0.23–0.30 s for one server alone.
• All 9 out-of-policy calls were refused (hidden write tools, another repository, path traversal, cloud metadata IP, loopback, file://, look-alike host). Without the gateway, the git server read a repository it was never meant to see.

What a gateway is for (and what it is not)

An MCP client connects to each server you configure and passes all their tools to the model. The servers usually don't know which agent is calling or why. A gateway adds that missing layer in one place:

• Tool scoping per role. A code-review agent gets git_status, git_diff and git_log, not git_reset. A tool the model never sees is a tool it can't be talked into calling.
• Argument checks. Paths inside an allowed folder, URLs on an allow-list. Tool arguments are written by the model, and the model reads untrusted text (web pages, issues, documents) that can steer what it writes.
• One audit log for every call by every agent, allowed or refused.
• Fewer tokens, because hidden tools are not in the context at all.

It is not a sandbox. If an allowed tool can do damage with allowed arguments, the gateway won't stop it. Use the servers' own restrictions too. As we found below, the reference servers already have some.

Setup

python -m venv .venv && source .venv/bin/activate
pip install "mcp>=2" tiktoken matplotlib
# the three official reference servers (Python versions), on PATH
pip install mcp-server-git mcp-server-fetch mcp-server-time

On mcp 2.x the high-level server class is MCPServer (the old FastMCP import fails; see our YOLO11 MCP server post). A gateway needs tool definitions that are only known at runtime, so it uses the low-level Server with on_list_tools and on_call_tool handlers instead of decorators.

The policy file

The policy names the upstream servers and, for each role, which tools it may see (glob patterns on server.tool), which it may never see, and which arguments are restricted. policy.json:

{
  "upstreams": {
    "time":  {"command": "mcp-server-time"},
    "fetch": {"command": "mcp-server-fetch"},
    "git":   {"command": "mcp-server-git"}
  },
  "roles": {
    "all": {"allow": ["*"]},
    "reviewer": {
      "allow": ["git.git_status", "git.git_diff*", "git.git_log", "git.git_show", "time.get_current_time"],
      "args": [{"tool": "git.*", "arg": "repo_path", "kind": "path", "allow": ["repo"]}]
    },
    "researcher": {
      "allow": ["fetch.fetch", "time.*"],
      "args": [{"tool": "fetch.fetch", "arg": "url", "kind": "url", "allow": ["example.com", "docs.python.org"]}]
    },
    "maintainer": {
      "allow": ["git.*", "fetch.*", "time.*"],
      "deny": ["git.git_reset", "git.git_checkout"],
      "args": [
        {"tool": "git.*", "arg": "repo_path", "kind": "path", "allow": ["repo"]},
        {"tool": "fetch.fetch", "arg": "url", "kind": "url", "allow": ["example.com", "docs.python.org"]}
      ]
    }
  }
}

Deny wins over allow, so maintainer gets all git tools except the two that throw work away. The all role is the baseline: everything, no checks.

The gateway

mcp_gateway.py (complete):

import fnmatch
import ipaddress
import json
import os
import sys
import time
from contextlib import AsyncExitStack, asynccontextmanager
from pathlib import Path
from urllib.parse import urlsplit

import anyio
import mcp_types as types
from mcp import Client, StdioServerParameters
from mcp.server.lowlevel import Server
from mcp.server.stdio import stdio_server

SEP = "__"  # exposed name = <server>__<tool>; dots are not accepted by every client


class Denied(Exception):
    """A policy refusal. Its message goes back to the model as a tool error."""


def check_path(value, allowed_dirs):
    real = os.path.realpath(value)
    if not any(real == d or real.startswith(d + os.sep) for d in map(os.path.realpath, allowed_dirs)):
        raise Denied(f"path not allowed: {value}")


def check_url(value, allowed_hosts):
    parts = urlsplit(value)
    host = (parts.hostname or "").lower()
    if parts.scheme not in ("http", "https") or not host:
        raise Denied(f"only http(s) URLs are allowed: {value}")
    try:  # literal IPs: never loopback, link-local (cloud metadata) or private ranges
        ip = ipaddress.ip_address(host)
        if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved:
            raise Denied(f"address not allowed: {host}")
    except ValueError:
        pass
    if not any(host == h or host.endswith("." + h) for h in allowed_hosts):
        raise Denied(f"host not on the allow-list: {host}")


ARG_CHECKS = {"path": check_path, "url": check_url}


class Gateway:
    def __init__(self, policy, role):
        self.upstreams = policy["upstreams"]
        self.role = policy["roles"][role]
        self.role_name = role
        self.clients = {}  # server name -> connected Client
        self.tools = {}    # exposed name -> (server, upstream tool name, Tool)

    def allowed(self, server, tool):
        name = f"{server}.{tool}"
        if any(fnmatch.fnmatchcase(name, p) for p in self.role.get("deny", [])):
            return False  # deny wins
        return any(fnmatch.fnmatchcase(name, p) for p in self.role.get("allow", []))

    async def start(self, stack):
        # start only the upstream servers this role can use at all
        wanted = {p.split(".")[0] for p in self.role.get("allow", [])}
        for server, cfg in self.upstreams.items():
            if server not in wanted and "*" not in wanted:
                continue
            params = StdioServerParameters(command=cfg["command"], args=cfg.get("args", []))
            client = await stack.enter_async_context(Client(params))
            self.clients[server] = client
            for tool in (await client.list_tools()).tools:
                if self.allowed(server, tool.name):
                    exposed = tool.model_copy(update={"name": f"{server}{SEP}{tool.name}"})
                    self.tools[exposed.name] = (server, tool.name, exposed)

    def check_args(self, server, tool, args):
        for rule in self.role.get("args", []):
            if not fnmatch.fnmatchcase(f"{server}.{tool}", rule["tool"]):
                continue
            value = args.get(rule["arg"])
            if value is None:
                continue
            ARG_CHECKS[rule["kind"]](str(value), rule["allow"])

    async def call(self, name, args):
        t0 = time.perf_counter()
        entry = self.tools.get(name)
        try:
            if entry is None:  # hidden and unknown tools look the same to the model
                raise Denied(f"unknown tool: {name}")
            server, tool, _ = entry
            self.check_args(server, tool, args)
        except Denied as e:
            audit(self.role_name, name, "denied", t0, str(e))
            return types.CallToolResult(content=[types.TextContent(text=f"Denied by gateway: {e}")], is_error=True)
        result = await self.clients[server].call_tool(tool, args)
        audit(self.role_name, name, "error" if result.is_error else "ok", t0)
        return result


def audit(role, tool, outcome, t0, reason=None):
    rec = {"role": role, "tool": tool, "outcome": outcome, "ms": round((time.perf_counter() - t0) * 1000, 2)}
    if reason:
        rec["reason"] = reason
    print(json.dumps(rec), file=sys.stderr, flush=True)  # stdout belongs to the protocol


def build_server(policy, role):
    gw = Gateway(policy, role)

    @asynccontextmanager
    async def lifespan(_server):
        async with AsyncExitStack() as stack:
            await gw.start(stack)
            yield gw

    async def list_tools(ctx, params):
        return types.ListToolsResult(tools=[t for _, _, t in ctx.lifespan_context.tools.values()])

    async def call_tool(ctx, params):
        return await ctx.lifespan_context.call(params.name, params.arguments or {})

    return Server(f"gateway-{role}", lifespan=lifespan, on_list_tools=list_tools, on_call_tool=call_tool)


async def main(policy_path):
    policy = json.loads(Path(policy_path).read_text())
    server = build_server(policy, os.environ["GATEWAY_ROLE"])
    async with stdio_server() as (read, write):
        await server.run(read, write, server.create_initialization_options())


if __name__ == "__main__":
    anyio.run(main, sys.argv[1])

The choices that matter:

• Filter at listing time and again at call time. The tool list is built once from the policy, and call only accepts names from that list. A client can send tools/call for any name, listed or not, so hiding alone is not enough.
• Hidden and nonexistent tools get the same answer (unknown tool), so a model probing names learns nothing.
• Refusals are tool errors, not protocol errors. is_error=True with a reason lets the model correct itself ("path not allowed") instead of failing the whole turn.
• realpath before comparing paths, so repo/../other-repo and symlinks don't slip through.
• Upstreams start in the lifespan and stay connected for the session. Only the servers a role can use are started.
• Audit to stderr. In a stdio server, stdout carries the protocol. One stray print() there breaks the session.

To use it, point your MCP client at the gateway instead of the individual servers, one entry per agent role:

{"mcpServers": {"reviewer-tools": {
  "command": "python", "args": ["mcp_gateway.py", "policy.json"],
  "env": {"GATEWAY_ROLE": "reviewer"}}}}

What it saves: tokens

We counted each tool definition (name, description, input and output schema, as JSON) with tiktoken's o200k_base encoding, then the full list each role sees through the gateway.

Two things surprised us when we looked per tool:

• Tool count is a poor proxy for cost. The time server has 3 tools and costs 1,295 tokens. The git server has 14 and costs 2,198. That's why the 4-tool researcher role (fetch + time) costs more than the 7-tool reviewer.
• Descriptions are a small part; schemas are most of it. Across all 18 tools the description strings were 297 of 3,883 tokens (8%). Output schemas were 591 tokens, all from the time server, where they are 46% of its cost. convert_time alone is 631 tokens, 379 of them its output schema with nested $defs. So "write shorter descriptions" won't move the number much. Remove tools the agent doesn't need.

The gateway's server__ prefixes added 36 tokens to the full list (3,883 → 3,919). Some clients now load tool definitions lazily or behind a search step, which changes how much of this is paid up front. Removing tools still helps in every client, because a hidden tool is never loaded or chosen.

What it costs: latency and startup

bench_gateway.py (below) timed warm calls after 5 warm-up calls: 200 calls of get_current_time and 100 of git_status on a one-commit repository, direct and through the gateway. Medians, range across three runs:

The gateway is a second stdio hop: one more JSON-RPC parse and serialise, one more pipe each way. That costs about 0.57 ms per call. The path check itself is too small to separate from noise; refused calls were decided in 0.2 ms or less in the audit log. For tools that do real work this is negligible. It would only matter for a tool that returns in microseconds and is called thousands of times per task.

Startup is different. Time to spawn and complete the first tools/list (median of 5 per run):

Our gateway starts upstreams one after another, so startup is roughly the sum of the servers plus the gateway's own Python start. Starting only what a role needs already saves about 250 ms. If startup matters, start the upstreams concurrently or run the gateway as a long-lived HTTP service shared by agents instead of one stdio process per session.

Does the policy hold?

We sent ten calls through the gateway: one legitimate read and nine that a manipulated model might try.

Then we checked what happens without the gateway, and the answer was more nuanced than "everything gets through":

• mcp-server-git started without options read other-repo and returned its commit history. Started with --repository repo, it refused: Repository path 'other-repo' is outside the allowed repository. The restriction exists, but only if you pass the flag.
• mcp-server-fetch already refuses private addresses by default. It resolves the host first (127.0.0.1 resolves to 127.0.0.1, which is not a public address), which is stronger than our gateway's check of literal IPs. It also has an --allowed-hosts option.

So configure the servers' own guards first. The gateway adds the parts a single server can't know about: which agent is calling, one policy across all servers, tools hidden per role, and one log. Our URL check only looks at literal IPs, so a hostname that resolves to a private address would pass it. Here the fetch server's own resolver catches that. If you put the gateway in front of servers without such guards, resolve hostnames in the gateway too.

Checklist for your own gateway

• Give each agent role its own tool list. Start from read-only tools and add write tools on purpose.
• Enforce at call time, not only in the listing. Answer hidden and unknown tools the same way.
• Check arguments the model writes: realpath for paths, scheme + host allow-list for URLs.
• Return refusals as tool errors with a reason the model can act on.
• Turn on the servers' own restrictions too (mcp-server-git --repository, mcp-server-fetch --allowed-hosts).
• Count your tool tokens per role. Schemas, not descriptions, are usually the bulk.
• Log every call, allowed and refused, to stderr or a file, never stdout.

Benchmark script

"""usage: python bench_gateway.py policy.json OUT_JSON
Needs the three reference servers on PATH, a git repo at ./repo and another at ./other-repo."""
import asyncio, json, os, platform, statistics, sys, time
from pathlib import Path

import tiktoken
from mcp import Client, StdioServerParameters

ENC = tiktoken.get_encoding("o200k_base")
GATEWAY = str(Path(__file__).with_name("mcp_gateway.py"))
COLD, WARMUP = 5, 5


def tokens(tool):
    return len(ENC.encode(json.dumps(tool.model_dump(mode="json", by_alias=True, exclude_none=True))))


def stats(xs):
    return {"median": round(statistics.median(xs), 3), "p90": round(sorted(xs)[int(0.9 * len(xs)) - 1], 3)}


def gateway_params(policy, role):
    return StdioServerParameters(command=sys.executable, args=[GATEWAY, policy],
                                 env={**os.environ, "GATEWAY_ROLE": role})


async def cold_start(params):
    out = []
    for _ in range(COLD):
        t0 = time.perf_counter()
        async with Client(params) as c:
            await c.list_tools()
            out.append((time.perf_counter() - t0) * 1000)
    return stats(out)


async def timed_calls(client, name, args, n):
    for _ in range(WARMUP):
        await client.call_tool(name, args)
    xs = []
    for _ in range(n):
        t0 = time.perf_counter()
        r = await client.call_tool(name, args)
        xs.append((time.perf_counter() - t0) * 1000)
        assert not r.is_error, r.content
    return stats(xs)


async def main(policy, out_json):
    pol = json.loads(Path(policy).read_text())
    report = {"servers": {}, "roles": {}, "latency_ms": {}}
    for name, cfg in pol["upstreams"].items():
        params = StdioServerParameters(command=cfg["command"])
        async with Client(params) as c:
            tools = (await c.list_tools()).tools
        report["servers"][name] = {"cold_start_ms": await cold_start(params), "tools": len(tools),
                                   "tokens": sum(tokens(t) for t in tools)}
    for role in pol["roles"]:
        params = gateway_params(policy, role)
        async with Client(params) as c:
            tools = (await c.list_tools()).tools
        report["roles"][role] = {"cold_start_ms": await cold_start(params), "tools": len(tools),
                                 "tokens": sum(tokens(t) for t in tools)}
    lat = report["latency_ms"]
    async with Client(StdioServerParameters(command="mcp-server-time")) as c:
        await c.list_tools()
        lat["time_direct"] = await timed_calls(c, "get_current_time", {"timezone": "UTC"}, 200)
    async with Client(StdioServerParameters(command="mcp-server-git")) as c:
        await c.list_tools()
        lat["git_status_direct"] = await timed_calls(c, "git_status", {"repo_path": "repo"}, 100)
    async with Client(gateway_params(policy, "all")) as c:
        await c.list_tools()
        lat["time_gateway"] = await timed_calls(c, "time__get_current_time", {"timezone": "UTC"}, 200)
        lat["git_status_gateway"] = await timed_calls(c, "git__git_status", {"repo_path": "repo"}, 100)
    Path(out_json).write_text(json.dumps(report, indent=1))


if __name__ == "__main__":
    asyncio.run(main(*sys.argv[1:3]))

We shortened the script for the page. The version we ran also times the reviewer role, runs the ten policy cases, and repeats the other-repo read without the gateway. A separate schema_split.py counts description and output-schema tokens per tool.

References

1. Model Context Protocol specification and documentation.
2. MCP Python SDK (version 2.3.0 used here).
3. MCP reference servers: mcp-server-git, mcp-server-fetch, mcp-server-time (version 2026.11.4 used here).
4. tiktoken (o200k_base encoding) for token counts.