MCP security best practices start from one fact: every tool call that reaches an MCP server comes from a client you should not trust, often an AI agent acting on text it read somewhere else. This guide is for developers who run MCP servers that agents call. It explains the MCP threat model from the official specification, turns it into a checklist, and then applies the checklist to one server step by step with Promptise Foundry. At the end you have a hardened server and a script that checks its controls.
How do you secure an MCP server?
Treat every tool call as an untrusted request. Authenticate each call and accept only tokens issued for this server. Give each tool the narrowest scope or role, and validate every argument with a tight schema. Put a person in front of irreversible actions, limit each client's rate, keep secrets out of tool output, and write a tamper-evident audit log. Lock down the transport, and prefer stdio for local servers. Then test the controls the way an attacker would.
Key takeaways
Bind tokens to the server. The spec says an MCP server must not accept tokens that were not issued for it, and it must not pass a client's token through to other APIs.
Authorization is per tool, not per server. A support agent that can read orders should not be able to refund them.
Most damage comes through normal tool calls. A valid caller with too much access, or a tool that returns a password, needs no exploit.
Order matters. Authenticate first, so the rate limiter and the audit log know who called.
[02]
What is the MCP threat model?
An MCP server sits between an AI application and the systems it can change. A user talks to a host application, the host runs an agent and an MCP client, and the client calls tools on one or more servers. The server then reads databases or calls other APIs with its own credentials. An authorization server issues the tokens.
user ──▶ host app + agent ──▶ MCP client ──(token)──▶ MCP server ──▶ database / APIs
│ │
│ reads tool results │ verifies: who is calling,
│ (untrusted text) │ which tools, which arguments
▼ ▼
prompt injection risk confused deputy, token passthrough,
over-broad scopes, leaked secretsEach arrow is a trust boundary. The agent can be tricked by text it reads, so a valid, authenticated client can still ask for something harmful. Unlike a normal API, you defend against your own legitimate callers as well as outsiders.
The official MCP Security Best Practices page for the current spec version (2026-07-28) names these attacks:
Confused deputy. An MCP proxy server that uses one static OAuth client ID for a third-party API can be tricked into sending an authorization code to an attacker, because a consent cookie skips the consent screen. Proxy servers must ask for consent per client.
Token passthrough. A server that accepts tokens issued for other services, or forwards a client's token downstream, breaks audit trails and lets a stolen token travel. Servers MUST NOT accept tokens that were not explicitly issued for them.
Server-side request forgery (SSRF). A malicious server puts internal URLs, such as a cloud metadata endpoint, in its OAuth metadata, and a client that follows them fetches internal resources. This is mainly a client-side duty.
State handle hijacking. The 2026-07-28 spec dropped protocol-level sessions, so servers that keep state hand out handles, such as a workflow ID. A handle must never count as authentication: bind it to the verified user. The 2025-11-25 page calls this session hijacking and applies the same rule to session IDs.
Local MCP server compromise. A server on a user's machine runs with the user's rights and can be reached by other local processes, and by web pages through DNS rebinding. The spec recommends stdio, or a token if you use HTTP.
Scope minimization. A token with broad scopes such as db:* turns one leak into access to everything. Start small and ask for more scopes only when a privileged tool needs them.
The page also covers client-side risks such as malicious authorization URLs and mix-up attacks. Two more risks come from how agents use MCP: prompt injection through tool results, and third-party servers that change their tool descriptions after you review them (tool poisoning).
[03]
What should an MCP security checklist include?
The table turns the threat model into controls you can check off. Points 1 to 11 are server-side and the steps below implement them. Points 12 to 14 belong to the agent and your process.
# | Control | Stops | In Promptise Foundry | Deep dive |
|---|---|---|---|---|
1 | Authenticate every call | Anonymous callers, session or handle reuse | MCPServer(require_auth=True) + AuthMiddleware | |
2 | Accept only short-lived tokens issued for this server | Token passthrough, replay on other servers | JWTAuth(audience=...), JwksAuth(audience=...); exp is required | |
3 | Give each tool the narrowest scope or role | Privilege escalation, over-broad tokens | guards=[HasScope(...)], roles=[...] | |
4 | List only the tools a caller may use | Reconnaissance, wasted agent turns | MCPServer(hide_unauthorized_tools=True) | This guide |
5 | Validate every argument with a tight schema | Injection-shaped input, absurd values | | |
6 | Return errors that leak nothing | Reconnaissance through stack traces | ToolError; other exceptions become INTERNAL_ERROR | This guide |
7 | Keep secrets out of code, config and tool output | Credential leaks | Environment variables + CredentialDetector redaction | This guide |
8 | Require human approval for irreversible tools | An agent, or a hijacked one, moving money | requires_approval=True + ApprovalGateMiddleware | |
9 | Limit rate, concurrency and time per client | Runaway agents, floods | RateLimitMiddleware, ConcurrencyLimiter, TimeoutMiddleware | |
10 | Keep a tamper-evident audit log, and verify it | Silent misuse, edited evidence | AuditMiddleware + promptise audit verify | |
11 | Lock down the transport | DNS rebinding, network exposure | Loopback bind, allowed_hosts, allowed_origins; stdio for local use | |
12 | Treat tool results as untrusted in the agent | Prompt injection through data | PromptiseSecurityScanner(scan_tool_results=True) | |
13 | Review and pin third-party servers | Tool poisoning, rug pulls | Not built in: see Limitations | This guide |
14 | Test the controls in CI | Regressions | TestClient tests + a check script |
[04]
When to apply which controls (and when not to)
Size the controls to who can reach the server and what its tools can change.
Server | Must have | Can skip |
|---|---|---|
Local stdio server for one desktop client | Input validation, secrets out of output, review of what it can touch | Tokens, rate limits: only the parent process can talk to it |
Local HTTP server | Everything for stdio, plus a token and Host checks | Network rate limits, if one client uses it |
Internal HTTP server for your own agents | Points 1 to 11 | Nothing, because agents can be injected |
Public or multi-tenant server | Points 1 to 14, OAuth with an identity provider, tenant isolation | Nothing |
Apply approvals with care. Gate irreversible or expensive tools, not every write, or reviewers start approving without reading.
[05]
What you need
Python 3.10 or later and Promptise Foundry 1.4.1 or later.
No model and no API key. The checks call the server directly, as any MCP client could.
A free local port. The examples use 18751.
About 25 minutes.
pip install "promptise>=1.4.1"PyJWT and httpx come with Promptise's dependencies. The check script uses PyJWT to mint and forge tokens.
The example is an orders server with four tools: get_order, search_orders, refund_order and integration_status, which returns the server's back-end configuration. Each step adds one part of the checklist to secure_server.py. The output lines in each step come from check_controls.py, the test script in step 7, run once against the finished server.
[06]
Step 1: Authenticate every call and bind tokens to this server
Authentication tells the server who is calling, and every later control depends on it. Audience binding is the spec's rule against token passthrough: a token minted for another service is refused even if it carries a valid signature.
server = MCPServer("orders", require_auth=True, hide_unauthorized_tools=True)
# Middleware runs in the order it is added: the first one is the outermost layer.
server.add_middleware(AuthMiddleware(JWTAuth(secret=JWT_SECRET, audience="orders-mcp")))No credentials refused 0.1s 157 chars
MCPConnectionRejectedError: Server at http://127.0.0.1:18751/mcp rejected the connection: 401 Un
Forged token (wrong key) refused 0.0s 157 chars
MCPConnectionRejectedError: Server at http://127.0.0.1:18751/mcp rejected the connection: 401 Un
Token minted for another server refused 0.0s 157 chars
MCPConnectionRejectedError: Server at http://127.0.0.1:18751/mcp rejected the connection: 401 Un
Expired token refused 0.0s 157 chars
MCPConnectionRejectedError: Server at http://127.0.0.1:18751/mcp rejected the connection: 401 Un
Valid token, allowed tool ok 0.0s 76 chars
{"order_id": "A-1001", "customer": "acme", "total": 129.0, "status": "paid"}What happened:
The secrets come from the environment. The server reads ORDERS_JWT_SECRET, ORDERS_DB_URL and PROMPTISE_AUDIT_SECRET with os.environ[...], so a missing one stops it at start-up instead of running with a default key.
`require_auth=True` makes every tool authenticate. Over HTTP, Promptise checks the token on every request and answers 401 before the MCP layer runs. It also tags the request with the token's principal, and the SDK transport compares that with the principal that created the session, so a session ID alone gets nobody in.
`audience="orders-mcp"` rejects the token minted for `billing-mcp`, although it is signed with the same key. JWTAuth also rejects tokens without an exp claim and expired tokens.
JWTAuth uses a shared HS256 key. With an identity provider, use JwksAuth.from_discovery(issuer=..., audience=...), which requires an audience. MCP server authentication covers API keys, JWTs and OAuth client credentials, and AI agent identity covers which token an agent should present.
[07]
Step 2: Give each tool the narrowest scope or role
A token proves who called. Guards decide what that caller may do, tool by tool: reads, refunds and operations data each need their own permission.
@server.tool(read_only_hint=True, guards=[HasScope("orders:read")])
async def get_order(order_id: OrderId) -> dict:@server.tool(destructive_hint=True, guards=[HasScope("refunds:write")], requires_approval=True)
async def refund_order(order_id: OrderId, amount: Amount) -> dict:@server.tool(read_only_hint=True, roles=["ops"])
async def integration_status() -> dict:Read-only agent issues a refund refused 0.0s 203 chars
{ "error": { "code": "ACCESS_DENIED", "message": "Requires any of scopes [refunds:write], but cl
Support agent reads ops config refused 0.0s 192 chars
{ "error": { "code": "ACCESS_DENIED", "message": "Requires any of roles [ops], but client has [(HasScope reads the JWT's space-separated scope claim, and roles=[...] reads its roles claim. The refusal names the missing permission.
hide_unauthorized_tools=True from step 1 runs the same guards on tools/list, so each caller sees only what it may call. list_tools.py connects with four tokens and prints the tool names each caller gets:
support-bot sees ['get_order', 'search_orders']
refund-agent sees ['get_order', 'refund_order', 'search_orders']
ops-agent sees ['get_order', 'integration_status', 'search_orders']
finance-lead sees ['approvals_decide', 'approvals_list']Hiding is least privilege for discovery, not the security boundary: a hidden tool called by name is still refused by its guards. It keeps tool names private and saves agents wasted turns. The approvals_list and approvals_decide tools come from step 4 and only finance-lead, who holds the approver role, sees them.
[08]
Step 3: Validate input and clean up output
Arguments come from a model, and a model can be steered by text it read. Tight schemas stop malformed and absurd values before your code runs. On the way out, errors and results must not carry secrets.
OrderId = Annotated[str, Field(pattern=r"^[A-Z]-\d{4}$", description="An order ID like A-1001.")]
Query = Annotated[str, Field(min_length=2, max_length=100)]
Amount = Annotated[float, Field(gt=0, le=500)]class RedactSecrets:
"""Middleware: scrub credentials from every string a tool returns."""
def __init__(self) -> None:
self.scanner = PromptiseSecurityScanner(detectors=[CredentialDetector()])
async def clean(self, value):
if isinstance(value, str):
return await self.scanner.check_output(value)
if isinstance(value, dict):
return {k: await self.clean(v) for k, v in value.items()}
if isinstance(value, list):
return [await self.clean(v) for v in value]
return value
async def __call__(self, ctx: RequestContext, call_next):
return await self.clean(await call_next(ctx))Refund larger than any order refused 0.0s 315 chars
{ "error": { "code": "VALIDATION_ERROR", "message": "Invalid input: amount: Input should be less
Path-traversal-shaped ID refused 0.0s 325 chars
{ "error": { "code": "VALIDATION_ERROR", "message": "Invalid input: order_id: String should matc
1 MB search string refused 0.0s 313 chars
{ "error": { "code": "VALIDATION_ERROR", "message": "Invalid input: query: String should have at
Unknown order ID refused 0.0s 88 chars
{ "error": { "code": "TOOL_ERROR", "message": "No order Z-9999.", "retryable": false } }
Tool output contains a DB password ok 0.0s 65 chars
{"ok": true, "config": {"warehouse_db": "[POSTGRES_CONNECTION]"}}What each piece does:
The schema becomes the tool's JSON schema and Promptise validates every call against it. A refund over 500, an ID that isn't A-1234, and a 1 MB search all get a short VALIDATION_ERROR. Validation runs inside the middleware chain, so rejected calls are rate limited and audited like any other.
`ToolError` sends a clean message to the caller. Any other exception a tool raises becomes INTERNAL_ERROR with a generic message, so a database URL in an exception never reaches the client.
`RedactSecrets` scans every string in the result with CredentialDetector, which recognizes common API keys, tokens and connection strings. The ops caller still gets the configuration, with the password replaced by [POSTGRES_CONNECTION].
Validation does not stop prompt injection. Injection arrives in the content a tool returns, such as a ticket or a web page, and targets the model that reads it. That is the agent's job, covered under "Running it in production".
[09]
Step 4: Require human approval for irreversible tools
A refund can't be undone, and the agent asking for it may have been tricked. requires_approval=True from step 2 parks the call until a person decides.
approver = PendingApprover(server, approver_role="approver")
server.add_middleware(ApprovalGateMiddleware(approver, timeout=5))Refund, nobody approves refused 5.0s 228 chars
{ "error": { "code": "APPROVAL_DENIED", "message": "Approval for tool 'refund_order' timed out a
reviewer finance-lead approved refund_order {'order_id': 'A-1001', 'amount': 129.0} from refund-agent
Refund, a person approves ok 0.2s 41 chars
{"order_id": "A-1001", "refunded": 129.0}PendingApprover registers two tools, approvals_list and approvals_decide, guarded by the approver role. The first refund got no decision in 5 seconds and was denied: the gate fails closed. For the second, finance-lead listed the pending call with its arguments and caller, and approved it. Nobody can approve their own call, even with the approver role. The gate also checks the tool's guards before it asks anyone, so a caller without refunds:write never reaches a reviewer.
Five seconds suits a demo. Use minutes in production, and add the gate before TimeoutMiddleware, or the tool timeout cuts the human wait short. Human-in-the-loop approval covers approvals from the agent's side.
[10]
Step 5: Limit each client's rate, concurrency and time
An agent in a loop can call a tool hundreds of times a minute. Limits per client protect your back end and your other callers.
server.add_middleware(RateLimitMiddleware(rate_per_minute=60, burst=10))server.add_middleware(ConcurrencyLimiter(max_concurrent=20))
server.add_middleware(TimeoutMiddleware(default_timeout=10.0))50 parallel calls from one client 10/50 servedbusy-bot sent 50 calls at once. The token bucket allowed a burst of 10 and refused the other 40 with RATE_LIMIT_EXCEEDED and a retry hint. The bucket is per client because AuthMiddleware runs first and sets the caller's ID. Add the rate limiter before authentication and every caller shares one global bucket. ConcurrencyLimiter caps calls in flight across the server, and TimeoutMiddleware ends calls that hang. MCP rate limiting per client covers per-tool limits, tiers and how an agent should back off.
[11]
Step 6: Keep a tamper-evident audit log
When an agent does something wrong, you need to know which caller asked for what, with which arguments, and whether the record was changed afterwards.
server.add_middleware(AuditMiddleware(log_path="audit.jsonl", include_args=True))Each entry holds the verified caller, the tool, the arguments and the outcome, HMAC-signed with PROMPTISE_AUDIT_SECRET and chained to the entry before it. Verify the log with the CLI, then change one recorded refund amount in a copy with a short script, tamper.py, and verify again:
$ promptise audit verify audit.jsonl
OK 63 entries: intact.
last_hash 0788f02d1ac908832c55a0ba7940df27b3cce82bd09d11b3f73aac13d4cb4fbe (seq
62)
exit=0
$ python tamper.py
edited entry seq=1: amount -> 1.0
$ promptise audit verify audit.jsonl
TAMPERED audit.jsonl:2: modified: the entry was changed after it was signed (its
signature no longer matches, the next entry still links to it)
FAILED 63 entries: 1 problem; the first is at audit.jsonl:2.
exit=1The edit is caught at the exact line, and the command exits with 1, so a cron job can alert on it. Denied calls, validation errors and approvals are all in the log with the caller's identity. The 401s from step 1 are not, because they never reach the MCP layer, so keep HTTP access logs too.
include_args=True records arguments, which may contain personal data. Turn it on only where your retention policy allows it. AI agent audit logging covers anchoring the end of the chain and shipping closed files to storage nobody can rewrite.
[12]
Step 7: Verify your controls with a check script
Write down the calls a careless or hostile client would make, and run them against your server. The script mints its own tokens, including forged and expired ones, so it needs only the server's test key:
"""Check that an MCP server's security controls engage: make the calls a careless
or hostile client would, and print what the server did with each one.
python check_controls.py http://127.0.0.1:18751/mcp
"""
import asyncio
import json
import os
import sys
import time
import httpx
import jwt
from promptise.mcp.client import MCPClient
URL = sys.argv[1]
SECRET = os.environ["ORDERS_JWT_SECRET"]
def token(scopes="", roles=(), aud="orders-mcp", secret=SECRET, ttl=300, sub="support-bot"):
now = int(time.time())
claims = {"sub": sub, "aud": aud, "scope": scopes, "roles": list(roles),
"iat": now, "exp": now + ttl}
return jwt.encode(claims, secret, algorithm="HS256")
SUPPORT = token("orders:read")
REFUNDER = token("orders:read refunds:write", sub="refund-agent")
OPS = token("orders:read", roles=["ops"], sub="ops-agent")
APPROVER = token(roles=["approver"], sub="finance-lead")
async def call(tool, args, bearer=None):
"""Return (outcome, text): 'ok' when the tool ran, else 'refused' and why."""
try:
async with MCPClient(url=URL, bearer_token=bearer) as client:
result = await client.call_tool(tool, args)
except Exception as exc:
return "refused", f"{type(exc).__name__}: {exc}"
text = " ".join(" ".join(getattr(c, "text", "").split()) for c in result.content)
return ("refused" if result.isError else "ok"), text
async def approve_next_refund():
"""A person with the approver role reviews the pending call and approves it."""
async with MCPClient(url=URL, bearer_token=APPROVER) as client:
for _ in range(20):
listed = await client.call_tool("approvals_list", {})
pending = json.loads(listed.content[0].text) if listed.content else []
if pending:
req = pending[0]
await client.call_tool("approvals_decide", {
"request_id": req["request_id"], "approve": True, "reason": "matches ticket"})
return f"finance-lead approved {req['tool']} {req['arguments']} from {req['client_id']}"
await asyncio.sleep(0.2)
return "nothing pending"
async def flood(n=50):
busy = token("orders:read", sub="busy-bot")
async with MCPClient(url=URL, bearer_token=busy) as client:
results = await asyncio.gather(*(client.call_tool("get_order", {"order_id": "A-1001"})
for _ in range(n)))
return sum(not r.isError for r in results), n
async def rebinding(bearer=None):
"""A web page whose DNS name was rebound to 127.0.0.1 sends its own Host header."""
body = {"jsonrpc": "2.0", "id": 1, "method": "initialize",
"params": {"protocolVersion": "2025-11-25", "capabilities": {},
"clientInfo": {"name": "evil-page", "version": "1"}}}
async with httpx.AsyncClient() as http:
r = await http.post(URL, json=body, headers={
"Host": "attacker.example", "Origin": "http://attacker.example",
"Accept": "application/json, text/event-stream",
**({"Authorization": f"Bearer {bearer}"} if bearer else {})})
return r.status_code
CHECKS = [
("No credentials", "get_order", {"order_id": "A-1001"}, None),
("Forged token (wrong key)", "get_order", {"order_id": "A-1001"},
token("orders:read", secret="a-guessed-secret-of-32-bytes-len")),
("Token minted for another server", "get_order", {"order_id": "A-1001"},
token("orders:read", aud="billing-mcp")),
("Expired token", "get_order", {"order_id": "A-1001"}, token("orders:read", ttl=-60)),
("Valid token, allowed tool", "get_order", {"order_id": "A-1001"}, SUPPORT),
("Read-only agent issues a refund", "refund_order", {"order_id": "A-1001", "amount": 129}, SUPPORT),
("Support agent reads ops config", "integration_status", {}, SUPPORT),
("Refund larger than any order", "refund_order", {"order_id": "A-1001", "amount": 1e9}, REFUNDER),
("Path-traversal-shaped ID", "get_order", {"order_id": "../../etc/passwd"}, SUPPORT),
("1 MB search string", "search_orders", {"query": "x" * 1_000_000}, SUPPORT),
("Unknown order ID", "get_order", {"order_id": "Z-9999"}, SUPPORT),
("Tool output contains a DB password", "integration_status", {}, OPS),
("Refund, nobody approves", "refund_order", {"order_id": "A-1001", "amount": 129}, REFUNDER),
]
def show(name, outcome, seconds, text):
leak = " SECRET VISIBLE" if "S3cr3t-pw" in text else ""
print(f"{name:<36} {outcome:<8} {seconds:4.1f}s {len(text):>6,} chars{leak}\n {text[:96]}")
async def main():
for name, tool, args, bearer in CHECKS:
started = time.perf_counter()
outcome, text = await call(tool, args, bearer)
show(name, outcome, time.perf_counter() - started, text)
started = time.perf_counter()
refund = asyncio.create_task(call("refund_order", {"order_id": "A-1001", "amount": 129}, REFUNDER))
print(f"{' reviewer':<36} {await approve_next_refund()}")
outcome, text = await refund
show("Refund, a person approves", outcome, time.perf_counter() - started, text)
served, n = await flood()
print(f"{'50 parallel calls from one client':<36} {served}/{n} served")
print(f"{'DNS rebinding, no token':<36} HTTP {await rebinding()}")
print(f"{'DNS rebinding, valid token':<36} HTTP {await rebinding(SUPPORT)}")
asyncio.run(main())Run it against the server from steps 1 to 6:
No credentials refused 0.1s 157 chars
MCPConnectionRejectedError: Server at http://127.0.0.1:18751/mcp rejected the connection: 401 Un
Forged token (wrong key) refused 0.0s 157 chars
MCPConnectionRejectedError: Server at http://127.0.0.1:18751/mcp rejected the connection: 401 Un
Token minted for another server refused 0.0s 157 chars
MCPConnectionRejectedError: Server at http://127.0.0.1:18751/mcp rejected the connection: 401 Un
Expired token refused 0.0s 157 chars
MCPConnectionRejectedError: Server at http://127.0.0.1:18751/mcp rejected the connection: 401 Un
Valid token, allowed tool ok 0.0s 76 chars
{"order_id": "A-1001", "customer": "acme", "total": 129.0, "status": "paid"}
Read-only agent issues a refund refused 0.0s 203 chars
{ "error": { "code": "ACCESS_DENIED", "message": "Requires any of scopes [refunds:write], but cl
Support agent reads ops config refused 0.0s 192 chars
{ "error": { "code": "ACCESS_DENIED", "message": "Requires any of roles [ops], but client has [(
Refund larger than any order refused 0.0s 315 chars
{ "error": { "code": "VALIDATION_ERROR", "message": "Invalid input: amount: Input should be less
Path-traversal-shaped ID refused 0.0s 325 chars
{ "error": { "code": "VALIDATION_ERROR", "message": "Invalid input: order_id: String should matc
1 MB search string refused 0.0s 313 chars
{ "error": { "code": "VALIDATION_ERROR", "message": "Invalid input: query: String should have at
Unknown order ID refused 0.0s 88 chars
{ "error": { "code": "TOOL_ERROR", "message": "No order Z-9999.", "retryable": false } }
Tool output contains a DB password ok 0.0s 65 chars
{"ok": true, "config": {"warehouse_db": "[POSTGRES_CONNECTION]"}}
Refund, nobody approves refused 5.0s 228 chars
{ "error": { "code": "APPROVAL_DENIED", "message": "Approval for tool 'refund_order' timed out a
reviewer finance-lead approved refund_order {'order_id': 'A-1001', 'amount': 129.0} from refund-agent
Refund, a person approves ok 0.2s 41 chars
{"order_id": "A-1001", "refunded": 129.0}
50 parallel calls from one client 10/50 served
DNS rebinding, no token HTTP 401
DNS rebinding, valid token HTTP 421Apart from the one allowed call, every line is a control engaging: four 401s, two ACCESS_DENIEDs, three VALIDATION_ERRORs, a clean TOOL_ERROR, a redacted password, a denied and an approved refund, a rate limit, and the transport checks covered below. Run the same list in CI with Promptise's TestClient, which needs no port. Testing MCP servers shows how.
[13]
The complete example
The full server, with every step in place:
"""An orders MCP server with the MCP security checklist applied.
Secrets come from the environment; the server refuses to start without them.
"""
import os
from typing import Annotated
from promptise.guardrails import CredentialDetector, PromptiseSecurityScanner
from promptise.mcp.server import (
ApprovalGateMiddleware,
AuditMiddleware,
AuthMiddleware,
ConcurrencyLimiter,
HasScope,
JWTAuth,
MCPServer,
PendingApprover,
RateLimitMiddleware,
RequestContext,
TimeoutMiddleware,
ToolError,
)
from pydantic import Field
# Secrets come from the environment: a missing one is a KeyError at start-up.
JWT_SECRET = os.environ["ORDERS_JWT_SECRET"]
DB_URL = os.environ["ORDERS_DB_URL"]
os.environ["PROMPTISE_AUDIT_SECRET"] # the audit key; AuditMiddleware reads it itself
ORDERS = {"A-1001": {"customer": "acme", "total": 129.00, "status": "paid"}}
# A tight schema for every argument.
OrderId = Annotated[str, Field(pattern=r"^[A-Z]-\d{4}$", description="An order ID like A-1001.")]
Query = Annotated[str, Field(min_length=2, max_length=100)]
Amount = Annotated[float, Field(gt=0, le=500)]
class RedactSecrets:
"""Middleware: scrub credentials from every string a tool returns."""
def __init__(self) -> None:
self.scanner = PromptiseSecurityScanner(detectors=[CredentialDetector()])
async def clean(self, value):
if isinstance(value, str):
return await self.scanner.check_output(value)
if isinstance(value, dict):
return {k: await self.clean(v) for k, v in value.items()}
if isinstance(value, list):
return [await self.clean(v) for v in value]
return value
async def __call__(self, ctx: RequestContext, call_next):
return await self.clean(await call_next(ctx))
server = MCPServer("orders", require_auth=True, hide_unauthorized_tools=True)
# Middleware runs in the order it is added: the first one is the outermost layer.
server.add_middleware(AuthMiddleware(JWTAuth(secret=JWT_SECRET, audience="orders-mcp")))
server.add_middleware(AuditMiddleware(log_path="audit.jsonl", include_args=True))
server.add_middleware(RateLimitMiddleware(rate_per_minute=60, burst=10))
server.add_middleware(RedactSecrets())
approver = PendingApprover(server, approver_role="approver")
server.add_middleware(ApprovalGateMiddleware(approver, timeout=5))
server.add_middleware(ConcurrencyLimiter(max_concurrent=20))
server.add_middleware(TimeoutMiddleware(default_timeout=10.0))
@server.tool(read_only_hint=True, guards=[HasScope("orders:read")])
async def get_order(order_id: OrderId) -> dict:
"""Get an order by ID."""
if order_id not in ORDERS:
raise ToolError(f"No order {order_id}.")
return {"order_id": order_id, **ORDERS[order_id]}
@server.tool(read_only_hint=True, guards=[HasScope("orders:read")])
async def search_orders(query: Query) -> list[dict]:
"""Search orders by customer name."""
return [{"order_id": k, **v} for k, v in ORDERS.items() if query.lower() in v["customer"]]
@server.tool(destructive_hint=True, guards=[HasScope("refunds:write")], requires_approval=True)
async def refund_order(order_id: OrderId, amount: Amount) -> dict:
"""Refund an order. Needs the refunds:write scope and a human approval."""
return {"order_id": order_id, "refunded": amount}
@server.tool(read_only_hint=True, roles=["ops"])
async def integration_status() -> dict:
"""Show which back-end systems this server talks to."""
return {"ok": True, "config": {"warehouse_db": DB_URL}}
if __name__ == "__main__":
server.run(transport="http", host="127.0.0.1", port=18751)mcp-security/
├── secure_server.py # the hardened orders server (steps 1-6)
├── check_controls.py # calls a hostile client would make (step 7)
├── list_tools.py # what each caller sees in tools/list (step 2)
├── tamper.py # edits one audit entry (step 6)
├── env.sh # generates demo secrets once
└── run_demo.sh # runs the commands below and stops the serverpip install "promptise>=1.4.1"
source env.sh # generates demo secrets once; use a secret manager in production
python secure_server.py &
python check_controls.py http://127.0.0.1:18751/mcp
python list_tools.py http://127.0.0.1:18751/mcp
promptise audit verify audit.jsonl[14]
Running it in production
Lock down the transport
A server bound to 127.0.0.1 can't be reached from the network, but a web page in your browser can reach it through DNS rebinding. Promptise checks Host and Origin on loopback binds by default. On any other bind it can't guess your public hostname, so validation is off until you name it. transport_policy.py asks build_transport_security, the internal function run() uses, which policy each bind gets:
127.0.0.1 -> on
0.0.0.0 -> off
0.0.0.0 + allowed_hosts -> ['mcp.example.com']In a container, call server.run(transport="http", host="0.0.0.0", allowed_hosts=["mcp.example.com"]) and terminate TLS at your proxy. The last two lines of the check output show both layers: a rebinding request without a token got 401, and with a stolen valid token the Host check still answered 421. For a server only one local client uses, prefer stdio: no port at all. See MCP transports and the deployment docs.
Keep secrets in a secret manager
In production, inject the three secrets from your secret manager, give each environment its own keys, and rotate them on a schedule. Old audit files verify only with the key that signed them, so keep retired audit keys.
Treat tool results as untrusted in the agent
Server controls can't stop a ticket or a web page that tells the agent to refund everything. Scan tool results before the model reads them:
from promptise import build_agent
from promptise.guardrails import PromptiseSecurityScanner
scanner = PromptiseSecurityScanner.default(scan_tool_results=True)
agent = await build_agent(model="openai:gpt-5-mini", servers=servers, guardrails=scanner)PromptiseSecurityScanner.default() uses a local injection model, so it needs pip install "promptise[all]". Pair it with approvals on the tools an injected agent could misuse. Prompt injection in AI agents and the guardrails docs cover the details.
Isolate tenants
A server for several customers must keep their data apart. Put the tenant in the token, check it with a guard such as HasTenant, and scope every query by it. Rate limit buckets already include the tenant when the token carries one, and PendingApprover(max_pending_per_tenant=...) caps pending approvals per tenant. See multi-tenant AI agents.
Review the servers your agents connect to
Read a third-party server's tool descriptions before you connect, pin the version you reviewed, and diff the tool list when it changes.
[15]
Common problems and fixes
Symptom | Cause | Fix |
|---|---|---|
A token from another service of yours works on this server | JWTAuth has no audience | Set audience= to this server's name, and mint tokens per server |
Every client shares one rate-limit bucket | RateLimitMiddleware is added before AuthMiddleware | Add AuthMiddleware first |
Approvals end with TIMEOUT before anyone can decide | TimeoutMiddleware wraps the approval gate | Add ApprovalGateMiddleware before TimeoutMiddleware |
A caller sees tools it can't use | hide_unauthorized_tools is off | Pass hide_unauthorized_tools=True to MCPServer |
ValueError at start-up about a signed audit log | No audit key is set | Set PROMPTISE_AUDIT_SECRET from your secret manager |
A browser page or proxy gets | The Host header is not in the allowed list | Add your public hostname to allowed_hosts |
Pending approvals disappear after a deploy | The approval store is in memory | Expect it: a restart denies pending calls, so callers must retry |
Invalid calls come back as a long plain-text Input validation error and are missing from the audit log | Promptise older than 1.3.4, where the MCP SDK validated arguments before the pipeline | Upgrade to 1.4.1 or later |
Failed logins are missing from the audit log | 401s are answered before the MCP layer | Keep HTTP access logs alongside the audit log |
[16]
Limitations
No OAuth discovery for clients. Promptise validates JWTs, API keys and JWKS-signed tokens, but it doesn't publish the RFC 9728 protected-resource metadata or send WWW-Authenticate scope challenges. Clients that expect the spec's discovery and step-up flow can't sign in on their own. Issue tokens out of band, as the authentication guide shows.
A stolen token works until it expires. Promptise keeps no list of used token IDs (jti), so a captured token replays freely on the server it was issued for. The audience check stops it everywhere else. Keep lifetimes in minutes.
Rate limits and approvals live in one process. Several replicas each keep their own buckets and pending approvals, and a restart drops pending approvals, which are then denied.
No pinning of third-party tools. Promptise scans tool results but doesn't hash or diff the tool descriptions a server sends, so a changed description is not flagged.
No consent flow for OAuth proxies. A server that proxies a third-party OAuth API must build the spec's per-client consent page itself.
[17]
Frequently asked questions
What are the biggest MCP security risks?
For servers: unauthenticated callers, tokens accepted from other services (token passthrough), tools with broader access than needed, irreversible actions with no human check, and secrets in tool output. For agents: prompt injection through tool results and third-party servers that change their tools. The official MCP security best practices also cover confused deputies, SSRF and session hijacking.
Does a local MCP server need authentication?
A stdio server doesn't: only the client process that started it can talk to it. A server that listens on HTTP does, even on 127.0.0.1, because any local process can reach the port, and a web page can through DNS rebinding. Host validation stops the browser case, and a token stops the rest.
Should an MCP server pass the user's token to downstream APIs?
No. The MCP spec forbids token passthrough. A server must reject tokens that weren't issued for it, and call other APIs with its own credentials or a token exchanged for that API. Otherwise the downstream API trusts a token nobody validated for it, and its logs show the wrong caller.
How do I limit what an AI agent can do through MCP?
Give the agent a token with only the scopes its task needs, and put a guard on every tool that checks for that scope or a role. Hide the tools it can't use from its tool list, validate every argument, and require human approval for tools that can't be undone. Then the worst a tricked agent can do is limited to its own permissions.
How do I audit what an AI agent did through MCP?
Log every tool call on the server with the caller's verified identity, the arguments and the outcome, signed and chained so edits are detectable, and verify the chain regularly. Server-side logs cover every client, including ones you don't control.
Is input validation enough to stop prompt injection?
No. Validation stops malformed arguments from reaching your code. Prompt injection arrives in the content tools return, such as a web page or a ticket, and targets the model that reads it. Defend against it in the agent: scan tool results, limit what each agent may call, and put approval in front of risky tools.
[18]
Next steps
Set up API keys, JWTs and OAuth client credentials: MCP server authentication.
Decide which token each agent presents: AI agent identity and authentication.
Build the server features this checklist relies on: build a production MCP server in Python.
Turn the check script into tests: testing MCP servers.
Reference: authentication and security, approval gates, observability and audit and testing in the Promptise docs.