Step 1: Sign agents in with API keys
A customer mints one key per agent. The key carries the workspace and an access level. The server stores only a SHA-256 hash of the secret.
KeyStore.lookup() is the hook the auth provider calls on every request:
mcp_server/keys.pypython
def lookup(self, presented: str) -> dict[str, Any] | None: """APIKeyAuth(lookup=...) hook: the key's identity, or None to refuse it.""" key_id, _, secret = presented.removeprefix(PREFIX).partition(".") if not presented.startswith(PREFIX) or not secret: return None record = self._load().get(key_id) if record is None or record.get("revoked_at"): return None if not secrets.compare_digest(_hash(secret), record["secret_sha256"]): return None return { "client_id": record["client_id"], "tenant_id": record["workspace"], "roles": record["roles"], }
The returned tenant_id becomes ctx.client.tenant_id. The roles feed the guards. The server is built with require_tenant=True, so no tool runs without a key that names a workspace. It also sets hide_unauthorized_tools=True, so a reader key never sees the write tools.
Each tool states its role in the decorator. Reads need reader. Writes need editor. Deleting a project needs admin plus a person's approval:
mcp_server/server.pypython
@server.tool(roles=["admin"], requires_approval=True, destructive_hint=True, rate_limit="5/min")async def delete_project(ctx: RequestContext, project_id: ProjectId) -> dict[str, Any]: """Permanently delete a project with all its tasks and comments. A person in the workspace must approve it first; the call waits for that decision. Use archive_project unless the user asked to delete.""" project = await api.call(ws(ctx), "GET", f"/projects/{project_id}") await api.call(ws(ctx), "DELETE", f"/projects/{project_id}") return {"deleted": project_id, "name": project["name"]}
The docstring is the description the model reads. It says when to use the tool. It also names the safer tool to prefer.
Fields that identify a person come from the key. They never come from the agent:
mcp_server/server.pypython
# The author comes from the verified key, never from the agent.comment = await api.call( ws(ctx), "POST", f"/tasks/{task_id}/comments", json={"body": body, "author": f"agent:{ctx.client_id}"})
Step 3: Make deletions wait for a person
requires_approval=True only works with an ApprovalGateMiddleware installed. The server refuses to build without one. Here the handler is a PendingApprover. A gated call waits in its store until a reviewer decides. No decision before the timeout means the call is denied.
The SDK's stock reviewer tools see every pending call on the server. On a multi-tenant server, that would show Acme's pending deletions to Globex's reviewers. So this project registers its own two tools, scoped to the reviewer's workspace:
mcp_server/approvals.pypython
entry = next((p for p in approver.pending() if p["request_id"] == request_id), None)# Another workspace's request looks exactly like an unknown one.if entry is None or entry["tenant_id"] != ctx.client.tenant_id: raise ToolError(f"No pending request {request_id!r} in your workspace.", code="NOT_FOUND")if approve and entry["client_id"] == ctx.client_id: raise ToolError("You cannot approve a call you made yourself.", code="ACCESS_DENIED")resolved = approver.decide(request_id, approve, reviewer_id=ctx.client_id, reason=reason or None)
In the real run, the reviewer took 45 seconds to answer. The agent's call stayed open and finished once it was approved.
Step 4: Log every call, limit every key, report health
The middleware order matters. Audit goes first, so refused calls are logged too:
mcp_server/server.pypython
# Audit goes first so refused calls (bad key, wrong role, rate limited,# approval denied) are in the log too.server.add_middleware(audit)server.add_middleware(AuthMiddleware(APIKeyAuth(lookup=keys.lookup)))server.add_middleware( RateLimitMiddleware(TokenBucketLimiter(rate_per_minute=settings.rate_per_minute, burst=settings.rate_burst)))server.add_middleware(ApprovalGateMiddleware(approver, timeout=settings.approval_timeout))register_approval_tools(server, approver)
Each audit entry has the tool, the client id, the workspace, the roles, the arguments and the outcome. Each entry's HMAC covers the one before it. promptise run audit checks the chain. Edit one field and the check fails at that line.
RateLimitMiddleware runs after AuthMiddleware, so each key gets its own bucket.
A readiness check tells your load balancer when the API behind the server is down:
mcp_server/server.pypython
health = HealthCheck()health.add_check("taskboard_api", api.healthy, required_for_ready=True)health.register_resources(server)
Step 5: Revoke a key while the server runs
APIKeyAuth(lookup=...) calls lookup() on every request. The key store re-reads its file whenever the file changes. So a revocation from another process applies to the next request:
outputtext
$ promptise run keys revoke 2794e3c1 && promptise run agent-reader "List the active projects."
revoked 2794e3c1
The Taskboard MCP server refused this agent key (HTTP 401). It is unknown or was revoked.
In production, point lookup() at the table your admin UI writes.
Step 6: Build the customer's agent
This is the code a customer writes. It needs the server URL and an agent key:
customer_agent/agent.pypython
return await build_agent( model=model, # A Prompt object, rendered again on every run (see prompts.py). instructions=ops_assistant, # The key travels as the x-api-key header on every MCP request. servers={"taskboard": HTTPServerSpec(url=url, api_key=api_key)}, # A plain tool loop: look up, act, answer. The task needs nothing more. agent_pattern="react", max_agent_iterations=15, # agent.chat(..., session_id=...) keeps the conversation across runs. conversation_store=SQLiteConversationStore(str(DATA / "conversations.db")) if remember else None, observe=observe if observe is not None else observability(),)
The instructions are a Prompt built from blocks in customer_agent/prompts.py. One block states today's date. Without it, the model cannot work out "due this Friday". The conversation store lets a follow-up question pick up where the sweep left off. The observability config writes an HTML report and a JSON trace of every model call and tool call.
Connect any MCP client
The same key works in any MCP client. In Cursor, add this to .cursor/mcp.json:
jsonjson
{
"mcpServers": {
"taskboard": {
"url": "http://127.0.0.1:8080/mcp",
"headers": { "x-api-key": "<agent key>" }
}
}
}
In Claude Code:
terminalbash
claude mcp add taskboard --transport http http://127.0.0.1:8080/mcp --header "x-api-key: <agent key>"
The README also shows Claude Desktop, through the mcp-remote bridge. The rules stay the same in every client, because they live in the server.