envoys-mcp

Cryptographic caller identity for MCP over Streamable HTTP, using Envoys RFC 9421 HTTP Message Signatures.

MCP's auth story tells a server "this is a valid token." Envoys tells it "this is agent scout@…, provably, on every tool call." That closes MCP's per-caller identity gap — so a server can allowlist, audit, and rate-limit by stable agent identity, not a bearer secret.

Two framework-agnostic pieces that compose into a both-sides deployment:

Install

pip install envoys-mcp            # core (envoys + httpx)
pip install "envoys-mcp[examples]"  # + mcp + uvicorn to run the examples

Server — gate an MCP server by agent identity

from mcp.server.fastmcp import FastMCP
from envoys_mcp import EnvoysVerifyMiddleware

mcp = FastMCP("demo")

@mcp.tool()
def add(a: int, b: int) -> int:
    return a + b

# Wrap FastMCP's Streamable-HTTP ASGI app. Only allowlisted agents get through.
app = EnvoysVerifyMiddleware(
    mcp.streamable_http_app(),
    allowlist=["scout@your-handle.envoys.me"],   # omit to admit any verifiable signer
)
# serve `app` with uvicorn

Verified requests arrive at your app with the caller's identity on the ASGI scope:

scope["state"]["envoys"]  # -> {"address": "scout@your-handle.envoys.me", "keyid": "https://envoys.me/agents/..."}

Client — sign every MCP request

from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
from envoys import Envoys
from envoys_mcp import EnvoysAuth

agent = Envoys.from_env()  # ENVOYS_AGENT_KEY / ADDRESS / PUBLIC_KEY / PRIVATE_KEY

async with streamablehttp_client(url, auth=EnvoysAuth(agent)) as (read, write, _):
    async with ClientSession(read, write) as session:
        await session.initialize()
        await session.call_tool("add", {"a": 2, "b": 3})

EnvoysAuth signs every request the client makes — initialize, tools/call, the SSE GET, the session DELETE — with zero changes to the MCP client itself.

Authorize inside a tool

caller_identity(ctx) reads the verified identity from the FastMCP Context, so a tool can authorize per-call — not just at the gate:

from mcp.server.fastmcp import Context
from envoys_mcp import caller_identity

@mcp.tool()
def delete_widget(id: int, ctx: Context) -> str:
    who = caller_identity(ctx)          # {"address": "...", "keyid": "..."} or None
    if who is None or who["address"] not in ADMINS:
        raise PermissionError("not allowed")
    ...

Measure adoption

SigningStats is a drop-in audit callback that scores real signed traffic — verified requests × distinct agents, counted from request logs, not asserted:

from envoys_mcp import SigningStats

stats = SigningStats()
app = EnvoysVerifyMiddleware(mcp.streamable_http_app(), audit=stats)
# ... later ...
stats.snapshot()
# {'verified_requests': 7, 'rejected_requests': 1, 'distinct_identities': 1,
#  'by_address': {'scout@your-handle.envoys.me': 7}}

What gets enforced

On each request, the verifier checks (via the Envoys SDK):

• Signature — Ed25519 over @method, @path, content-digest (RFC 9421).
• Timestamp window — rejects stale/future-dated signatures (±300s / −30s).
• Replay — a given signature is accepted once.
• Content-Digest — body tampering is detected (digest is recomputed over the received body).
• Key pinning — first-seen public key per address is pinned; a silently rotated key fails until you Envoys.reset_pin(address).
• Allowlist — only listed agents are admitted (after the crypto check passes).

Key resolution

The verifier resolves each signer's public key from the keyid embedded in the signature (the agent's address URL, e.g. https://envoys.me/agents/scout@…). For agents registered on envoys.me this works with no configuration. For offline tests, monkeypatch Envoys.resolve_key_from_keyid (see tests/).

One gotcha: @path must match

The signed @path is the URL path only (no query). The client's request path and the server's scope["path"] must be byte-identical — so point the client at the exact MCP mount path (default /mcp) and avoid trailing-slash redirects.

Status

Verified two ways:

• 6 unit tests over an in-process ASGI round-trip — signed-passes, and unsigned / tampered-body / replayed / non-allowlisted all rejected, plus exempt-path bypass. No network or MCP SDK required.

  pip install "envoys-mcp[dev]" && pytest
  

• A live demo (examples/local_demo.py) — boots a real FastMCP Streamable-HTTP server + real MCP client (verified against mcp 1.27.2). Every request in the MCP lifecycle — initialize, the SSE GET, tools/call, DELETE — is signed and verified with live HTTP key resolution; an unsigned request is rejected with 401.

  pip install -e ".[examples]" && python examples/local_demo.py   # -> [demo] PASS
  

License

Apache-2.0