OAuth, API keys, secret storage, per-user state — and how to keep them out of your logs.
MCP Servers Module 3
Until now, PKB has been single-user and trust-everything. The moment you ship a server over HTTP (Lesson 13), or want per-user data, or call paid third-party APIs from your tools — auth becomes mandatory.
This lesson covers the integration points; full OAuth provider implementation is its own topic.
Authenticating the user to your server (OAuth for remote MCP, or local-user inference for stdio).
Authenticating your server to third-party APIs (API keys, service accounts, scoped tokens).
1. Client → /authorize (user grants consent)
2. Server → redirect with code
3. Client → /token (exchange code → access_token)
4. Client → MCP requests with `Authorization: Bearer <token>`
5. Server validates token per request
Standard OAuth 2.1 + PKCE; the MCP spec defines how the access token is attached to RPC frames.
MCP requires the server to be able to refuse unauthenticated requests and to scope state by the authenticated principal.
The official Python SDK exposes hooks for this; FastMCP wraps them. We won't implement a full provider — we'll stub one and show the integration points.
import os
from mcp.server.fastmcp import FastMCP
THIRD_PARTY_KEY = os.environ.get("PKB_THIRD_PARTY_KEY")
if not THIRD_PARTY_KEY:
raise SystemExit("PKB_THIRD_PARTY_KEY env var is required")
mcp = FastMCP("pkb", version="0.2.0")
Instead of a global NOTES dict, key state by the authenticated user:
NOTES[user_id][title] = body
The user id comes from the verified OAuth token (or from a stdio-context principal for local installs).
# pkb/auth.py
def get_current_user(ctx) -> str:
"""Return user id from ctx. Stub: reads from env, real impl reads OAuth claims."""
user = os.environ.get("PKB_USER", "local")
if not user:
raise McpError(ErrorData(code=INVALID_PARAMS, message="not authenticated"))
return user
This is the integration point — swap the stub for a real OAuth claim extractor when going remote.
pkb.export tool@mcp.tool()
def export(ctx: Context) -> dict[str, str]:
"""Export all of the current user's notes."""
user = get_current_user(ctx)
return USER_NOTES[user]
Gated by get_current_user; raises if no user — no auth bypass.
.env files (gitignored) + python-dotenv.
Vault / AWS Secrets Manager / GCP Secret Manager / 1Password Connect.
Never commit secrets; never echo them in tool responses.
Credential leakage: tools that return raw HTTP responses can echo Authorization headers; tools that error with str(exception) can include the API key in the stack trace; logs at debug level often dump request bodies including tokens. Audit all three paths.
pkb/auth.py with the stub get_current_user.NOTES to be per-user (USER_NOTES[user]).pkb.export() gated by auth.PKB_USER is unset.