=> ({
content: [{ type: "text", text: JSON.stringify(freshData) }],
}));
```
**Correct -- React app using `@modelcontextprotocol/ext-apps/react`:**
```tsx
import { useToolResult } from "@modelcontextprotocol/ext-apps/react";
function Dashboard() {
const result = useToolResult(); // receives tool call data
const data = JSON.parse(result?.content?.[0]?.text ?? "{}");
return {/* render interactive UI from data */}
;
}
```
**Key rules:**
- Use `ui://` URI scheme for all UI resources, with `text/html;profile=mcp-app` mimeType
- Use `registerAppTool` and `registerAppResource` from `@modelcontextprotocol/ext-apps/server`
- Link tools to UIs via `_meta.ui.resourceUri` on the tool definition
- Declare CSP domains explicitly: `connectDomains` (fetch), `resourceDomains` (CDN), `frameDomains` (iframes)
- Omitting CSP defaults to `connect-src 'none'` -- no external network access
- Set `visibility: ["app"]` for tools only the UI should call (hides from model)
- Default visibility is `["model", "app"]` -- tool visible to both model and UI
- Host renders UI in sandboxed iframe; never assume permissions are granted
- Content MUST be valid HTML5 provided via `text` (string) or `blob` (base64)
Reference: [MCP Apps Extension (SEP-1865)](https://github.com/modelcontextprotocol/ext-apps)
### OAuth 2.1 Authorization for MCP Servers — HIGH
## OAuth 2.1 Authorization for MCP Servers
MCP servers are OAuth 2.1 Resource Servers (spec 2025-11-25). Clients MUST use PKCE with S256, bind tokens to the target resource via RFC 8707, and never pass tokens through to downstream services.
**Incorrect -- no PKCE, no resource indicator, token passthrough:**
```typescript
// BAD: Missing PKCE and resource parameter
const authUrl = `${authServer}/authorize?client_id=${clientId}&redirect_uri=${redirect}`;
// BAD: Passing client's token to upstream API (confused deputy)
async function callUpstreamApi(clientToken: string) {
return fetch("https://api.example.com/data", {
headers: { Authorization: `Bearer ${clientToken}` }, // NEVER DO THIS
});
}
// BAD: No audience validation on the resource server
function validateToken(token: string) {
const decoded = jwt.verify(token, publicKey);
return decoded; // Missing audience check — accepts ANY valid token
}
```
**Correct -- PKCE S256 + RFC 8707 resource binding:**
```typescript
import crypto from "node:crypto";
// 1. PKCE: Generate verifier and S256 challenge
function createPkce() {
const verifier = crypto.randomBytes(32).toString("base64url");
const challenge = crypto.createHash("sha256").update(verifier).digest("base64url");
return { verifier, challenge };
}
// 2. Authorization request with resource indicator (RFC 8707)
function buildAuthUrl(
authServer: string, clientId: string, redirectUri: string,
mcpServerUri: string, scopes: string[],
) {
const { verifier, challenge } = createPkce();
const state = crypto.randomBytes(16).toString("base64url");
const params = new URLSearchParams({
response_type: "code",
client_id: clientId,
redirect_uri: redirectUri,
code_challenge: challenge,
code_challenge_method: "S256",
resource: mcpServerUri, // MUST match MCP server's canonical URI
scope: scopes.join(" "),
state,
});
return { url: `${authServer}/authorize?${params}`, verifier, state };
}
// 3. Token exchange — resource parameter MUST match authorization request
async function exchangeCode(
tokenEndpoint: string, code: string, verifier: string,
clientId: string, redirectUri: string, mcpServerUri: string,
) {
const res = await fetch(tokenEndpoint, {
method: "POST",
headers: { "Content-Type": "application/x-www-form-urlencoded" },
body: new URLSearchParams({
grant_type: "authorization_code", code,
code_verifier: verifier, client_id: clientId,
redirect_uri: redirectUri, resource: mcpServerUri,
}),
});
return res.json();
}
```
**Correct -- token validation + confused deputy prevention:**
```typescript
// 4. MCP server validates audience (RFC 8707 + RFC 9068)
function validateAccessToken(token: string, expectedAudience: string) {
const decoded = jwt.verify(token, publicKey, {
algorithms: ["RS256"],
audience: expectedAudience, // MUST be this server's canonical URI
issuer: trustedIssuer,
});
return decoded;
}
// 5. Upstream calls use a SEPARATE token — never forward the client's token
async function callUpstream(upstreamTokenEndpoint: string) {
const { access_token } = await fetch(upstreamTokenEndpoint, {
method: "POST",
body: new URLSearchParams({ grant_type: "client_credentials", scope: "upstream:read" }),
}).then((r) => r.json());
return access_token; // Scoped to upstream, NOT the client's token
}
```
**Correct -- discovery, registration, and incremental scope consent:**
```typescript
// 6. Protected Resource Metadata discovery (RFC 9728)
async function discoverAuthServer(mcpServerUrl: string) {
const origin = new URL(mcpServerUrl).origin;
const meta = await fetch(`${origin}/.well-known/oauth-protected-resource`).then((r) => r.json());
const asUrl = meta.authorization_servers[0];
// Try OAuth 2.0 AS Metadata, then OIDC Discovery
for (const p of ["/.well-known/oauth-authorization-server", "/.well-known/openid-configuration"]) {
const res = await fetch(`${asUrl}${p}`);
if (res.ok) return res.json();
}
throw new Error("No authorization server metadata found");
}
// 7. Dynamic Client Registration (RFC 7591) — fallback when no pre-registration
async function registerClient(registrationEndpoint: string) {
return fetch(registrationEndpoint, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
client_name: "My MCP Client",
redirect_uris: ["http://127.0.0.1:3000/callback"],
grant_types: ["authorization_code"],
token_endpoint_auth_method: "none",
}),
}).then((r) => r.json());
}
// 8. Incremental scope consent — handle 403 insufficient_scope
function handleInsufficientScope(wwwAuth: string) {
const match = wwwAuth.match(/scope="([^"]+)"/);
if (match) return match[1].split(" "); // Re-authorize with these scopes
}
```
**Correct -- OAuth URL paste fallback (headless/SSH environments):**
```typescript
// 9. When browser can't open (SSH, containers, headless), print URL for manual paste
async function authorizeWithFallback(authUrl: string) {
const canOpenBrowser = process.env.DISPLAY || process.platform === "darwin";
if (canOpenBrowser) {
await open(authUrl); // Opens default browser
} else {
// Fallback: print URL for user to paste in their local browser
console.log("\nOpen this URL in your browser to authorize:");
console.log(`\n ${authUrl}\n`);
console.log("After authorizing, paste the callback URL here:");
const callbackUrl = await readline.question("> ");
const code = new URL(callbackUrl).searchParams.get("code");
if (!code) throw new Error("No authorization code found in callback URL");
return code;
}
}
```
**Key rules:**
- PKCE with S256 is mandatory; refuse to proceed if AS lacks `code_challenge_methods_supported`
- Include `resource` parameter (RFC 8707) in both authorization and token requests, set to the MCP server's canonical URI
- MCP servers MUST validate the `aud` claim matches their own URI — reject all other tokens
- NEVER pass the client's access token to upstream APIs (confused deputy); obtain a separate token via client credentials or token exchange (RFC 8693)
- Use Protected Resource Metadata (RFC 9728) for AS discovery; support both OAuth 2.0 AS Metadata and OIDC Discovery
- Prefer Client ID Metadata Documents over Dynamic Client Registration (RFC 7591) for new implementations
- Handle `403 insufficient_scope` by re-authorizing with scopes from the `WWW-Authenticate` header
- For high-security deployments, bind tokens to client certificates via mTLS (RFC 8705) to prevent token theft and replay
- In headless environments (SSH, containers, CI), implement URL paste fallback — print the auth URL for the user to open manually and accept the callback URL pasted back (see example 9 above)
### Implement MCP client patterns for reliable connections and multi-server orchestration — MEDIUM
## Client Patterns
Set up MCP clients with proper session management, error handling, and reconnection. Covers TypeScript and Python SDKs for consuming MCP servers from applications.
**Incorrect -- no error handling, no cleanup:**
```typescript
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
const client = new Client({ name: "app", version: "1.0.0" });
const transport = new StreamableHTTPClientTransport(new URL("http://localhost:3000/mcp"));
await client.connect(transport);
const result = await client.callTool({ name: "search", arguments: { q: "test" } });
console.log(result.content[0].text); // Crashes if tool errors or content empty
// Transport never closed -- connection leaked
```
```python
from mcp.client.streamable_http import streamable_http_client
from mcp import ClientSession
# No context manager -- session never cleaned up
read, write = await streamable_http_client("http://localhost:3000/mcp").__aenter__()
session = ClientSession(read, write)
await session.initialize()
result = await session.call_tool("search", arguments={"q": "test"})
print(result.content[0].text) # No type check, no error handling
```
**Correct -- TypeScript client with reconnection and capability negotiation:**
```typescript
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
const transport = new StreamableHTTPClientTransport(
new URL("http://localhost:3000/mcp"),
{
sessionId: cachedSessionId, // Reconnect to existing session
reconnectionOptions: {
maxRetries: 5,
initialReconnectionDelay: 1000,
maxReconnectionDelay: 30000,
reconnectionDelayGrowFactor: 1.5,
},
}
);
const client = new Client(
{ name: "my-app", version: "1.0.0" },
{ capabilities: { sampling: {} } } // Declare client capabilities
);
try {
await client.connect(transport);
const caps = client.getServerCapabilities();
// Discover tools before calling
const { tools } = await client.listTools();
const hasTool = tools.some((t) => t.name === "search");
if (!hasTool) throw new Error("Required tool 'search' not available");
const result = await client.callTool({ name: "search", arguments: { q: "test" } });
for (const content of result.content) {
if (content.type === "text") console.log(content.text);
}
} finally {
await transport.terminateSession();
await transport.close();
}
```
**Correct -- Python client with context managers:**
```python
import asyncio
from mcp import ClientSession, StdioServerParameters, types
from mcp.client.stdio import stdio_client
from mcp.client.streamable_http import streamable_http_client
async def run_stdio_client():
server_params = StdioServerParameters(
command="python", args=["my_server.py"]
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
tools = await session.list_tools()
result = await session.call_tool("add", arguments={"a": 5, "b": 3})
for content in result.content:
if isinstance(content, types.TextContent):
print(content.text)
async def run_http_client():
async with streamable_http_client("http://localhost:8000/mcp") as (read, write, get_session_id):
async with ClientSession(read, write) as session:
await session.initialize()
tools = await session.list_tools()
print([t.name for t in tools.tools])
```
**Correct -- multi-server orchestration (TypeScript):**
```typescript
async function connectServers(urls: string[]) {
const clients = await Promise.all(
urls.map(async (url) => {
const transport = new StreamableHTTPClientTransport(new URL(url));
const client = new Client({ name: "orchestrator", version: "1.0.0" });
await client.connect(transport);
const { tools } = await client.listTools();
return { client, transport, tools, url };
})
);
// Build unified tool registry across servers
const toolMap = new Map();
for (const entry of clients) {
for (const tool of entry.tools) {
toolMap.set(`${tool.name}@${entry.url}`, entry);
}
}
return { clients, toolMap };
}
```
**Key rules:**
- Always close transports in `finally` blocks (TS) or use context managers (Python)
- Call `initialize()` before any other session method in Python
- Discover tools with `listTools()` before calling -- never assume tool availability
- Use `reconnectionOptions` with exponential backoff for remote HTTP servers
- Cache `sessionId` to resume sessions after reconnection
- Check `content.type` before accessing `.text` -- tools may return images or errors
- For multi-server setups, namespace tools by server to avoid name collisions
- Declare client capabilities (`sampling`, `elicitation`) during construction
Reference: https://modelcontextprotocol.io/specification/2025-11-25/architecture
### Use MCP elicitation safely with consent handling and secure form-mode data collection — MEDIUM
## Elicitation
MCP elicitation lets servers request structured input from users at runtime via form mode (JSON Schema) or URL mode (external flows). Form mode collects non-sensitive data in-band; URL mode redirects users to secure pages for credentials, OAuth, or payments.
**Incorrect -- requesting secrets via form mode, ignoring decline/cancel:**
```python
@mcp.tool()
async def connect_api(ctx: Context) -> str:
# WRONG: form mode exposes secrets to the LLM context
result = await ctx.session.create_elicitation(
mode="form",
message="Enter your API key",
requestedSchema={
"type": "object",
"properties": {
"api_key": {"type": "string"},
# WRONG: nested objects not allowed in elicitation schemas
"config": {"type": "object", "properties": {"timeout": {"type": "number"}}},
},
},
)
# WRONG: assumes accept, crashes on decline/cancel
return call_api(result.content["api_key"])
```
**Correct -- form mode for non-sensitive data, flat schema, handle all actions:**
```python
@mcp.tool()
async def configure_search(ctx: Context) -> str:
result = await ctx.session.create_elicitation(
mode="form",
message="Configure your search preferences",
requestedSchema={
"type": "object",
"properties": {
"query": {"type": "string", "minLength": 1, "description": "Search terms"},
"category": {
"type": "string",
"enum": ["docs", "code", "issues"],
"default": "docs",
},
"max_results": {
"type": "integer",
"minimum": 1,
"maximum": 50,
"default": 10,
},
},
"required": ["query"],
},
)
if result.action == "accept":
return search(result.content)
elif result.action == "decline":
return "Search cancelled. Let me know if you'd like to try different options."
else: # cancel
return "Search dismissed. I can search with defaults if you'd like."
```
**Correct -- URL mode for sensitive data (API keys, OAuth):**
```python
@mcp.tool()
async def connect_service(ctx: Context) -> str:
elicitation_id = str(uuid.uuid4())
result = await ctx.session.create_elicitation(
mode="url",
message="Please authorize access to your account.",
elicitation_id=elicitation_id,
url=f"https://myserver.example.com/connect?eid={elicitation_id}",
)
if result.action == "accept":
# User consented to open URL -- interaction happens out-of-band.
# Server sends notifications/elicitation/complete when done.
return "Authorization started. I'll proceed once you complete the flow."
elif result.action == "decline":
return "Authorization declined. Some features will be unavailable."
else: # cancel
return "Authorization dismissed."
```
**Correct -- client declares elicitation capabilities:**
```typescript
const client = new Client({
name: "my-client",
version: "1.0.0",
}, {
capabilities: {
elicitation: { form: {}, url: {} }, // declare supported modes
},
});
```
**Key rules:**
- Never request secrets (API keys, passwords, tokens) via form mode -- use URL mode instead
- Schemas must be flat objects with primitive properties only (string, number, integer, boolean, enum) -- no nested objects or `$ref`
- Always handle all three response actions: `accept`, `decline`, `cancel`
- URL mode `accept` means user consented to open the URL, not that the flow is complete -- listen for `notifications/elicitation/complete`
- Clients must show the full URL and get explicit consent before opening; never auto-fetch or auto-navigate
- Servers must verify the user who completes a URL flow is the same user who initiated it (prevent phishing/account takeover)
- Check client capabilities before sending elicitation requests -- clients may support only `form`, only `url`, or both
Reference: https://modelcontextprotocol.io/specification/2025-11-25/client/elicitation
### Vet MCP servers from registries to prevent supply-chain attacks and data exfiltration — LOW
## Registry Discovery
Use the official MCP Registry API for programmatic server discovery and apply a vetting checklist before installing any third-party server.
**Incorrect -- blindly install unvetted servers:**
```python
# Grabbed a random server name from a blog post
config = {"mcpServers": {"sketchy-db": {"command": "npx", "args": ["@unknown/mcp-db"]}}}
# No source review, no version pinning, no permission audit
```
**Correct -- query the official registry and vet before installing:**
```python
import httpx
REGISTRY = "https://registry.modelcontextprotocol.io"
async def discover_servers(query: str) -> list[dict]:
"""Search the official MCP Registry API."""
async with httpx.AsyncClient() as client:
resp = await client.get(f"{REGISTRY}/v0.1/servers", params={
"search": query, "version": "latest", "limit": 20,
})
resp.raise_for_status()
return resp.json()["servers"]
async def get_server_detail(name: str, version: str = "latest") -> dict:
"""Fetch full metadata for a specific server."""
async with httpx.AsyncClient() as client:
resp = await client.get(f"{REGISTRY}/v0.1/servers/{name}/versions/{version}")
resp.raise_for_status()
return resp.json()
def vet_server(server: dict) -> list[str]:
"""Return warnings if server fails vetting checks."""
warnings = []
s = server.get("server", server)
if not s.get("repository", {}).get("url"):
warnings.append("No public source repository")
if not s.get("packages"):
warnings.append("Not published to any package registry")
meta = server.get("_meta", {}).get("io.modelcontextprotocol.registry/official", {})
if meta.get("status") != "active":
warnings.append(f"Registry status: {meta.get('status', 'unknown')}")
return warnings
```
**Community directories for broader discovery:**
| Directory | URL | Notes |
|-----------|-----|-------|
| Official Registry | registry.modelcontextprotocol.io | API-accessible, moderation |
| mcp.run | mcp.run | Hosted runtime, sandboxed |
| Smithery | smithery.ai | Install counts, reviews |
| Glama | glama.ai/mcp/servers | Curated catalog |
| MCP Servers | mcpservers.org | Community-maintained list |
**Vetting checklist before installing any server:**
```markdown
- [ ] Source code in a public repository with commit history
- [ ] Published to npm/PyPI (not just a git clone)
- [ ] Version pinned in config (no `@latest` in production)
- [ ] README documents all tools, resources, and required permissions
- [ ] No overly broad capabilities (filesystem root, network wildcard)
- [ ] Active maintenance (commits within last 90 days)
- [ ] Listed in official registry or reputable directory
```
**Icon metadata (spec 2025-11-25) -- expose icons for tools/resources:**
```python
@mcp.tool(metadata={"icon": "https://example.com/icons/search.svg"})
def search(query: str) -> str:
"""Search documents."""
...
```
**Key rules:**
- Always query the official registry at `registry.modelcontextprotocol.io/v0.1/servers` first
- Never install a server without checking its source repository and package provenance
- Pin exact versions in MCP server configurations -- avoid `@latest` in production
- Cross-reference multiple directories (registry, smithery, mcp.run) for trust signals
- Treat community servers as untrusted by default; apply allowlist patterns from security-hardening
- Use `vet_server()` checks programmatically when building multi-server orchestrations
### Bound MCP sampling loops with user approval to prevent unbounded LLM call chains — MEDIUM
## Sampling with Tool Calling
MCP sampling lets servers request LLM completions from clients, with optional tool definitions for agentic multi-turn loops. The client controls model access and user approval throughout.
**Incorrect -- no iteration cap, skips user approval:**
```python
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("agent-server")
@mcp.tool()
async def run_agent(task: str, ctx) -> str:
messages = [{"role": "user", "content": {"type": "text", "text": task}}]
tools = [{"name": "search", "description": "Search docs",
"inputSchema": {"type": "object", "properties": {"q": {"type": "string"}}, "required": ["q"]}}]
# Unbounded loop -- runs forever if LLM keeps calling tools
while True:
result = await ctx.session.create_message(
messages=messages, tools=tools, max_tokens=2000
)
if result.stop_reason != "toolUse":
return result.content.text
# Blindly append and continue without any limit
messages.append({"role": "assistant", "content": result.content})
tool_results = [execute_tool(tc) for tc in result.content]
messages.append({"role": "user", "content": tool_results})
```
**Correct -- bounded loop, tool choice control, proper message structure:**
```python
from mcp.server.fastmcp import FastMCP, Context
mcp = FastMCP("agent-server")
MAX_ITERATIONS = 5
TOOLS = [{
"name": "search",
"description": "Search documentation by keyword",
"inputSchema": {
"type": "object",
"properties": {"q": {"type": "string", "description": "Search query"}},
"required": ["q"],
},
}]
@mcp.tool()
async def run_agent(task: str, ctx: Context) -> str:
"""Run a bounded agent loop with tool access via sampling."""
messages = [{"role": "user", "content": {"type": "text", "text": task}}]
for i in range(MAX_ITERATIONS):
# Force text-only response on final iteration
tool_choice = (
{"mode": "none"} if i == MAX_ITERATIONS - 1
else {"mode": "auto"}
)
result = await ctx.session.create_message(
messages=messages,
tools=TOOLS,
tool_choice=tool_choice,
max_tokens=2000,
)
# LLM chose not to use tools -- return final answer
if result.stop_reason != "toolUse":
return result.content.text if hasattr(result.content, "text") else str(result.content)
# Execute each tool call, build tool_result messages
assistant_content = result.content if isinstance(result.content, list) else [result.content]
messages.append({"role": "assistant", "content": assistant_content})
# Tool results MUST be in their own user message -- no mixed content
tool_results = []
for block in assistant_content:
if block.type == "tool_use":
output = await execute_tool(block.name, block.input)
tool_results.append({
"type": "tool_result",
"toolUseId": block.id,
"content": [{"type": "text", "text": str(output)}],
})
messages.append({"role": "user", "content": tool_results})
return "Agent reached iteration limit without a final answer."
```
**Declaring sampling capability with tool support (client-side):**
```python
# Client must advertise sampling.tools capability during initialization
capabilities = {
"sampling": {
"tools": {} # Required for tool-enabled sampling requests
}
}
```
**Key rules:**
- Always cap iteration count and use `toolChoice: \{mode: "none"\}` on the final turn to force a text response
- Tool result messages MUST contain only `tool_result` blocks -- never mix with text or image content
- Every `tool_use` block (by `id`) must have a matching `tool_result` (by `toolUseId`) before the next assistant turn
- Clients MUST declare `sampling.tools` capability; servers MUST NOT send tool-enabled requests without it
- Human-in-the-loop: clients SHOULD present sampling requests and tool calls for user review before execution
- Use `toolChoice` modes: `auto` (LLM decides), `required` (must call a tool), `none` (text only)
- Parallel tool calls are supported -- handle arrays of `tool_use` blocks in a single assistant message
- Implement rate limiting on the client side to prevent runaway sampling loops
### Apply zero-trust verification to MCP servers to prevent rug-pull and data exfiltration — HIGH
## Security Hardening
Verify every tool with hash-based integrity checks. Use zero-trust allowlists, capability enforcement, and secure sessions.
**Incorrect -- trust all tools without verification:**
```python
tools = await mcp.list_tools() # No vetting!
result = await mcp.call_tool(name, args) # No integrity check!
session_id = f"{user_id}:{auth_token}" # CREDENTIAL LEAK in session ID!
```
**Correct -- zero-trust tool allowlist with hash verification:**
```python
from hashlib import sha256
from dataclasses import dataclass
from datetime import datetime, timezone
@dataclass
class AllowedTool:
name: str
description_hash: str
capabilities: list[str]
approved_by: str
max_calls_per_minute: int = 60
class MCPToolAllowlist:
def __init__(self):
self._allowed: dict[str, AllowedTool] = {}
self._call_counts: dict[str, list[datetime]] = {}
def register(self, tool: AllowedTool) -> None:
self._allowed[tool.name] = tool
self._call_counts[tool.name] = []
def validate(self, name: str, description: str) -> tuple[bool, str]:
if name not in self._allowed:
return False, f"Tool '{name}' not in allowlist"
expected = self._allowed[name]
actual_hash = sha256(description.encode('utf-8')).hexdigest()
if actual_hash != expected.description_hash:
return False, "Description changed (possible rug pull)"
# Rate limit
now = datetime.now(timezone.utc)
recent = [t for t in self._call_counts[name]
if (now - t).total_seconds() < 60]
if len(recent) >= expected.max_calls_per_minute:
return False, "Rate limit exceeded"
self._call_counts[name] = recent + [now]
return True, "OK"
```
**Correct -- capability enforcement (least privilege):**
```python
from enum import Enum
class ToolCapability(Enum):
READ_FILE = "read:file"
WRITE_FILE = "write:file"
EXECUTE_COMMAND = "execute:command"
NETWORK_REQUEST = "network:request"
SENSITIVE_PATHS = ["/etc/passwd", "~/.ssh", ".env", "credentials"]
class CapabilityEnforcer:
def __init__(self):
self._declarations: dict[str, set[ToolCapability]] = {}
def register(self, tool_name: str, caps: set[ToolCapability]) -> None:
self._declarations[tool_name] = caps
def check(self, tool_name: str, cap: ToolCapability, resource: str = "") -> tuple[bool, str]:
if tool_name not in self._declarations:
return False, "No capability declaration"
if cap not in self._declarations[tool_name]:
return False, f"Capability {cap.value} not allowed"
if cap in (ToolCapability.READ_FILE, ToolCapability.WRITE_FILE):
if any(s in resource for s in SENSITIVE_PATHS):
return False, "Sensitive path denied"
return True, "Allowed"
```
**Correct -- secure session management:**
```python
import secrets
def generate_session_id() -> str:
return secrets.token_urlsafe(32) # 256 bits of entropy
# NEVER: session_id = f"{user_id}:{auth_token}"
# ALWAYS: session_id = secrets.token_urlsafe(32)
```
**Rug pull detection -- hash comparison on every call:**
```python
class ToolIntegrityMonitor:
def __init__(self):
self._fingerprints: dict[str, str] = {}
def register(self, tool: dict) -> None:
desc = tool.get("description", "")
params = json.dumps(tool.get("parameters", {}), sort_keys=True)
combined = sha256(f"{desc}:{params}".encode()).hexdigest()
self._fingerprints[tool["name"]] = combined
def verify(self, tool: dict) -> tuple[bool, str | None]:
name = tool["name"]
if name not in self._fingerprints:
return False, "Tool not registered"
desc = tool.get("description", "")
params = json.dumps(tool.get("parameters", {}), sort_keys=True)
current = sha256(f"{desc}:{params}".encode()).hexdigest()
if current != self._fingerprints[name]:
return False, f"Tool '{name}' modified since registration"
return True, None
```
**Key rules:**
- Every tool must be explicitly vetted before use (zero-trust)
- Hash-verify description + parameters on every invocation
- Use `secrets.token_urlsafe(32)` for session IDs, never embed auth tokens
- Enforce least-privilege capabilities per tool
- Rate limit tool calls (per-tool and per-session)
- Auto-suspend tools that fail integrity checks
### Defend against prompt injection in MCP tool descriptions that can hijack LLM behavior — HIGH
## Security Injection Defense
Treat ALL tool descriptions as untrusted input. Normalize encodings, detect injection patterns, and sanitize before LLM exposure.
**Incorrect -- raw tool description passed to LLM:**
```python
# INJECTION RISK: description may contain "ignore previous instructions..."
prompt = f"Use this tool: {tool.description}"
tools = await mcp.list_tools() # No validation!
```
**Correct -- sanitize tool descriptions before use:**
```python
import re
FORBIDDEN_PATTERNS = {
"critical": [
(r"ignore\s+(all\s+)?previous", "instruction_override"),
(r"you\s+are\s+now", "role_hijack"),
(r"forget\s+(everything|all|above)", "context_wipe"),
(r"system\s*prompt", "system_access"),
],
"high": [
(r"IMPORTANT\s*:", "attention_hijack"),
(r"override\s+(all\s+)?settings", "config_override"),
(r"<\|.*?\|>", "delimiter_attack"),
(r"reveal\s+(your|the)\s+(prompt|instructions)", "prompt_extraction"),
],
}
def sanitize_description(description: str) -> tuple[str, list[str]]:
"""Sanitize tool description. Returns (sanitized, detected_threats)."""
if not description:
return "", []
threats = []
sanitized = normalize_encodings(description)
for level in ["critical", "high"]:
for pattern, name in FORBIDDEN_PATTERNS[level]:
if re.search(pattern, sanitized, re.I):
threats.append(f"{level}:{name}")
sanitized = re.sub(pattern, "[REDACTED]", sanitized, flags=re.I)
return sanitized.strip(), threats
```
**Correct -- normalize encodings to reveal hidden attacks:**
```python
import html
import urllib.parse
import unicodedata
HOMOGLYPHS = {
'\u0430': 'a', '\u0435': 'e', '\u043e': 'o',
'\u0440': 'p', '\u0441': 'c', '\u0443': 'y',
}
def normalize_encodings(text: str) -> str:
"""Decode HTML entities, URL encoding, hex escapes, homoglyphs."""
result = html.unescape(text) # I -> I
result = urllib.parse.unquote(result) # %69 -> i
result = re.sub( # \x69 -> i
r'\\x([0-9a-fA-F]{2})',
lambda m: chr(int(m.group(1), 16)),
result,
)
result = unicodedata.normalize('NFKC', result) # Unicode normalization
for glyph, latin in HOMOGLYPHS.items(): # Cyrillic -> Latin
result = result.replace(glyph, latin)
return result
```
**Correct -- filter sensitive data from tool responses:**
```python
RESPONSE_FILTERS = [
(r"api[_-]?key\s*[:=]\s*\S+", "[API_KEY_REDACTED]"),
(r"password\s*[:=]\s*\S+", "[PASSWORD_REDACTED]"),
(r"bearer\s+\S+", "[TOKEN_REDACTED]"),
(r"-----BEGIN.*KEY-----[\s\S]*-----END.*KEY-----", "[PRIVATE_KEY_REDACTED]"),
]
def filter_tool_response(response: str) -> str:
for pattern, replacement in RESPONSE_FILTERS:
response = re.sub(pattern, replacement, response, flags=re.I)
return response
```
**Key rules:**
- Always normalize encodings BEFORE pattern matching
- Block on critical threats (instruction override, role hijack)
- Redact high-severity patterns but allow the tool through
- Filter tool responses for secrets before they reach the LLM
- Test with known attack payloads: base64, homoglyphs, HTML entities
### Set up MCP servers with proper lifecycle management and structured error handling — HIGH
## Server Setup
Use FastMCP with lifespan context for shared resources. Define tools with explicit schemas and return errors as text content.
**Incorrect -- no lifecycle, raw exception:**
```python
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("my-server")
db = Database.connect() # Global -- never cleaned up
@mcp.tool()
def query(sql: str) -> str:
return db.query(sql) # Crashes on connection failure
```
**Correct -- FastMCP with lifespan and error handling:**
```python
from contextlib import asynccontextmanager
from collections.abc import AsyncIterator
from dataclasses import dataclass
from mcp.server.fastmcp import Context, FastMCP
@dataclass
class AppContext:
db: Database
cache: CacheService
@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
db = await Database.connect()
cache = await CacheService.connect()
try:
yield AppContext(db=db, cache=cache)
finally:
await cache.disconnect()
await db.disconnect()
mcp = FastMCP("my-server", lifespan=app_lifespan)
@mcp.tool()
def query(sql: str, ctx: Context) -> str:
"""Execute a read-only SQL query. Returns up to 100 rows."""
try:
app = ctx.request_context.lifespan_context
return app.db.query(sql)
except DatabaseError as e:
return f"Error: {e}" # Claude sees and can retry
```
**Tool definition best practices:**
```python
from mcp.types import Tool
Tool(
name="search_products",
description="Search product catalog. Returns up to 10 results.",
inputSchema={
"type": "object",
"properties": {
"query": {"type": "string", "description": "Search terms"},
"category": {
"type": "string",
"enum": ["electronics", "clothing", "books"],
},
"max_results": {
"type": "integer", "minimum": 1, "maximum": 50, "default": 10,
},
},
"required": ["query"],
},
)
```
**Key rules:**
- Always use lifespan for database connections, caches, HTTP clients
- Return errors as `TextContent` -- never raise unhandled exceptions
- Include `description` for every schema property
- Use `enum` for fixed option sets, `minimum`/`maximum` for numbers
- Use `asyncio.to_thread()` for blocking synchronous operations
- Limit response sizes (Claude has context limits). For intentionally large results (DB schemas, API specs), use `_meta["anthropic/maxResultSizeChars"]` annotation (up to 500K) so clients and hooks respect the declared size instead of truncating:
```python
@mcp.tool()
async def get_schema() -> dict:
schema = await db.get_full_schema()
return {
"_meta": {"anthropic/maxResultSizeChars": 100000},
"content": [{"type": "text", "text": schema}]
}
```
### Choose the right MCP transport for production multi-client and deployment requirements — HIGH
## Server Transport
Choose stdio for CLI/Desktop, Streamable HTTP for web apps and production multi-client. SSE is deprecated.
**Transport decision matrix:**
| Transport | Use Case | Pros | Cons |
|-----------|----------|------|------|
| stdio | CLI, Claude Desktop | Simple, no network | Single client only |
| SSE | **Deprecated** | Browser-compatible | Deprecated since March 2025 |
| Streamable HTTP | Web apps, production APIs | Multi-client, scalable, stateless option | More setup |
**Incorrect -- hardcoded transport, no configuration:**
```python
# Forces stdio -- can't switch to web deployment
from mcp.server.stdio import stdio_server
async def main():
async with stdio_server() as (read, write):
await server.run(read, write, server.create_initialization_options())
```
**Correct -- Python stdio server:**
```python
from mcp.server import Server
from mcp.server.stdio import stdio_server
server = Server("my-tools")
# Register handlers...
async def main():
async with stdio_server() as (read, write):
await server.run(read, write, server.create_initialization_options())
if __name__ == "__main__":
import asyncio
asyncio.run(main())
```
**Correct -- TypeScript stdio server:**
```typescript
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
const server = new Server(
{ name: "my-tools", version: "1.0.0" },
{ capabilities: { tools: {} } }
);
// Register handlers...
await server.connect(new StdioServerTransport());
```
**Deprecated -- SSE for web deployment (use Streamable HTTP instead):**
> SSE transport was deprecated in March 2025. Migrate to Streamable HTTP for
> new projects. SSE remains functional but receives no new features.
```python
from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.routing import Route
sse = SseServerTransport("/messages")
async def handle_sse(request):
async with sse.connect_sse(
request.scope, request.receive, request._send
) as streams:
await server.run(
streams[0], streams[1],
server.create_initialization_options()
)
app = Starlette(routes=[
Route("/sse", endpoint=handle_sse),
Route("/messages", endpoint=sse.handle_post_message, methods=["POST"]),
])
```
**Correct -- Streamable HTTP server (Python, recommended):**
```python
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("my-tools")
@mcp.tool()
def greet(name: str = "World") -> str:
"""Greet someone by name."""
return f"Hello, {name}!"
if __name__ == "__main__":
# Stateless with JSON responses -- best for production
mcp.run(transport="streamable-http", stateless_http=True, json_response=True)
# Stateful with session persistence (when needed):
# mcp.run(transport="streamable-http")
```
**Correct -- Streamable HTTP server (TypeScript, recommended):**
```typescript
import { createServer } from "node:http";
import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
const server = new McpServer({ name: "my-tools", version: "1.0.0" });
// Register handlers...
createServer(async (req, res) => {
const transport = new StreamableHTTPServerTransport({
sessionIdGenerator: undefined, // stateless; use () => randomUUID() for sessions
});
await server.connect(transport);
await transport.handleRequest(req, res);
}).listen(3000);
```
**Migrating SSE → Streamable HTTP:**
- Python: Replace `SseServerTransport` with `FastMCP.run(transport="streamable-http")`
- TypeScript: Replace `SSEServerTransport` with `StreamableHTTPServerTransport`
- Client endpoint changes from `/sse` + `/messages` to single `/mcp` path
- Streamable HTTP supports both stateless (scalable) and stateful (session) modes
**Claude Desktop configuration:**
```json
{
"mcpServers": {
"my-tools": {
"command": "npx",
"args": ["-y", "@myorg/my-tools"],
"env": { "DATABASE_URL": "postgres://..." }
},
"python-tools": {
"command": "uv",
"args": ["run", "python", "-m", "my_mcp_server"],
"cwd": "/path/to/project"
}
}
}
```
**Key rules:**
- Use Streamable HTTP for all new web/production deployments (SSE is deprecated)
- Use `uv` (not `pip`) for Python MCP server commands in Claude Desktop config
- Set `cwd` when the server needs access to project files
- Pass secrets via `env`, never hardcode in args
- TypeScript servers: use `npx -y` for zero-install execution
- Prefer stateless mode (`stateless_http=True`) unless session persistence is required
### Test and debug MCP servers to catch broken tools and transport failures before production — MEDIUM
## Testing & Debugging
Write automated tests for every tool using the SDK's in-process `Client`, and use MCP Inspector for interactive debugging of transports and auth.
**Incorrect -- manual testing only, no assertions:**
```python
# "I'll just test it in Claude Desktop"
mcp = FastMCP("my-server")
@mcp.tool()
def search(query: str) -> str:
return db.search(query)
# No tests, no fixtures, no CI -- bugs found by end users
```
**Correct -- unit tests with in-process Client:**
```python
import pytest
from fastmcp import Client # in-process harness lives in the separate `fastmcp` package
from mcp.types import CallToolResult, TextContent
from server import app
@pytest.fixture
def anyio_backend():
return "asyncio"
@pytest.fixture
async def client():
async with Client(app, raise_exceptions=True) as c:
yield c
@pytest.mark.anyio
async def test_search_returns_results(client: Client):
result = await client.call_tool("search", {"query": "test"})
assert isinstance(result, CallToolResult)
assert len(result.content) > 0
assert result.content[0].type == "text"
@pytest.mark.anyio
async def test_search_empty_query(client: Client):
result = await client.call_tool("search", {"query": ""})
assert "Error" in result.content[0].text # Graceful error, not crash
```
**Correct -- parametrized edge-case tests:**
```python
@pytest.mark.anyio
@pytest.mark.parametrize("args", [{"query": ""}, {"max_results": -1}, {}])
async def test_invalid_inputs_return_errors(client: Client, args):
result = await client.call_tool("search", args)
assert result.isError or "Error" in result.content[0].text
```
**Correct -- integration test with stdio transport:**
```python
import subprocess, json
def test_stdio_transport_connects():
"""Verify the server starts and responds to initialize over stdio."""
proc = subprocess.Popen(
["uv", "run", "server.py"],
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
)
init_msg = {"jsonrpc": "2.0", "id": 1, "method": "initialize",
"params": {"capabilities": {}, "clientInfo": {"name": "test"},
"protocolVersion": "2025-03-26"}}
proc.stdin.write(json.dumps(init_msg).encode() + b"\n")
proc.stdin.flush()
line = proc.stdout.readline()
assert b'"result"' in line # Server responded to init
proc.terminate()
```
**Interactive debugging with MCP Inspector:**
```bash
# Inspect a local Python server
npx @modelcontextprotocol/inspector uv run server.py
# Inspect a PyPI package
npx @modelcontextprotocol/inspector uvx mcp-server-git --repository ~/repo
# Inspect with environment variables
npx @modelcontextprotocol/inspector -e API_KEY=xxx uv run server.py
# Use Inspector to: list tools/resources, test tool calls with custom
# inputs, check capability negotiation, and view server logs.
# For scaffolding new servers, see the mcp-builder skill.
```
**Debug common connection failures:**
```bash
# Timeout: slow lifespan init blocks connection -- keep lifespan under 5s
# Auth 401: pass secrets via Inspector's -e flag or .env file
# "Connection refused": wrong transport -- match stdio vs Streamable HTTP
# Hang on tool call: blocking sync code -- wrap with asyncio.to_thread()
```
**Key rules:**
- Use `Client(app, raise_exceptions=True)` for unit tests -- no transport overhead
- Test both valid inputs and edge cases (empty, missing, out-of-range)
- Use `@pytest.mark.anyio` with `anyio_backend` fixture for async tests
- Use MCP Inspector (`npx @modelcontextprotocol/inspector`) for interactive debugging
- Keep lifespan initialization under 5s so Inspector and clients can connect
- Test stdio transport separately with `subprocess` for integration coverage
- Install test deps: `pip install inline-snapshot pytest anyio`
### Integrate WebMCP browser mediation correctly to avoid confusing it with standard MCP — LOW
## WebMCP Browser Integration
WebMCP is a W3C Community Group standard that exposes structured tools to AI agents inside the browser via `navigator.modelContext`. It complements MCP (not a replacement) — MCP handles AI-to-backend over JSON-RPC, WebMCP handles AI-to-browser-UI via in-page callbacks.
**Incorrect -- registering tools without input schema or user mediation:**
```typescript
// No schema, no description, no user interaction handling
navigator.modelContext.registerTool({
name: "submit-order",
description: "Submit order",
execute: async (input) => {
// Directly mutates state with no user confirmation
await fetch("/api/orders", { method: "POST", body: JSON.stringify(input) });
return { status: "submitted" };
},
});
```
**Correct -- full schema, annotations, and user interaction request:**
```typescript
navigator.modelContext.registerTool({
name: "submit-order",
description: "Submit the current shopping cart as an order. Requires user confirmation.",
inputSchema: {
type: "object",
properties: {
cartId: { type: "string", description: "Cart identifier" },
shipping: { type: "string", enum: ["standard", "express"] },
},
required: ["cartId"],
},
annotations: { readOnlyHint: false },
execute: async (input, client) => {
// Request explicit user confirmation before mutating state
const confirmed = await client.requestUserInteraction(async () => {
return window.confirm(`Place order for cart ${input.cartId}?`);
});
if (!confirmed) return { status: "cancelled_by_user" };
const res = await fetch("/api/orders", {
method: "POST",
body: JSON.stringify(input),
});
return { status: "submitted", orderId: (await res.json()).id };
},
});
```
**Read-only tool with annotations:**
```typescript
navigator.modelContext.registerTool({
name: "get-product-details",
description: "Retrieve product name, price, and availability from the current page.",
inputSchema: {
type: "object",
properties: {
productId: { type: "string", description: "Product ID visible on page" },
},
required: ["productId"],
},
annotations: { readOnlyHint: true },
execute: async (input) => {
const el = document.querySelector(`[data-product-id="${input.productId}"]`);
return el ? { name: el.dataset.name, price: el.dataset.price } : { error: "Not found" };
},
});
```
**When to use MCP vs WebMCP:**
| Concern | MCP | WebMCP |
|---------|-----|--------|
| Transport | JSON-RPC (stdio / SSE / HTTP) | In-page callbacks |
| Runs on | Server / backend | Browser (SecureContext) |
| Use case | DB queries, APIs, file I/O | DOM access, form fill, UI actions |
| Auth | OAuth 2.1 / tokens | Browser-mediated permission |
**Key rules:**
- WebMCP complements MCP — use MCP for backend services, WebMCP for browser-side UI tools
- Always provide `inputSchema` with property descriptions so agents understand parameters
- Set `annotations.readOnlyHint: true` on tools that only read data (no side effects)
- Use `client.requestUserInteraction()` before any state-mutating operation
- WebMCP requires `SecureContext` (HTTPS only) — `navigator.modelContext` is undefined on HTTP
- Call `unregisterTool(name)` or `clearContext()` during SPA route teardown to prevent stale tools
- Keep tool descriptions specific — agents select tools by description, not by probing
---
## References (2)
### Mcp Audit Runbook
# MCP Audit Runbook
Operational procedure for re-running the MCP version audit documented in `mcp-version-matrix.md`.
## Transport check (run this before the version check)
Version drift is the second question. The first is whether the server should be on npx at
all. For every entry using `command: npx`, check whether upstream publishes a hosted HTTP
endpoint; if it does, prefer it:
```json
"context7": { "type": "http", "url": "https://mcp.context7.com/mcp" }
```
stdio spawns **one child process per Claude Code session**, so its cost scales with how
many sessions the machine keeps open, not with how many projects are configured. Keep
stdio only where the hosted endpoint is unreachable (proxy, air gap) or does not exist.
## When to run
- **Calendar trigger:** every 90 days from the last audit's "Last audited" header
- **Event trigger:** when a consumer skill reports unexpected MCP behavior
- **Release trigger:** before every minor OrchestKit release (pin HIGH-tier servers if drift detected)
## Re-run script
```bash
# From repo root
for pkg in @upstash/context7-mcp @modelcontextprotocol/server-sequential-thinking \
@modelcontextprotocol/server-memory tavily-mcp \
@21st-dev/magic fal-ai-mcp notebooklm-mcp; do
v=$(npm view "$pkg" version 2>/dev/null)
pub=$(npm view "$pkg" time.modified 2>/dev/null | head -1)
printf "%-50s %-15s %s\n" "$pkg" "$v" "$pub"
done
```
## Interpreting output
Compare the script output to the "Upstream latest" column in `mcp-version-matrix.md`.
| Drift observed | Action |
|---|---|
| No change | Update "Last audited" header in matrix. Done. |
| Patch bump on any tier | Update matrix. No consumer-side action needed. |
| Minor bump on LOW/MEDIUM | Update matrix. Spot-check the package's release notes for regressions that affect current usage patterns. |
| Minor bump on HIGH | Read the full changelog. If breaking, file an issue to pin the previous version in `.mcp.json` and adapt consumer skills. |
| Major bump on any tier | Read the changelog. Treat as a potentially-breaking event regardless of tier. |
| Package unpublished / yanked | File urgent issue. Pin to last-known-good in `.mcp.json`. |
## Updating the matrix
After each audit:
1. Update the "Upstream latest" and "Last published" columns
2. Update the "Last audited" header with today's date
3. If tier classification changes (e.g., a 0.x server hits 1.0), update the Risk Tier table
4. Commit with message `chore: refresh MCP version audit matrix (\{YYYY-MM-DD\})`
## When to escalate to a full consumer audit
Trigger a consumer-side review (read every `mcp__<name>__*` call site) when any of:
- MCP upstream emits a major version bump
- Upstream changelog mentions "breaking" or "removed"
- A skill's MCP-using phase starts failing in production after the fetch
- npm publishes a new package under the same scope (namespace squat check)
## Escape hatch: pin a specific version
Pinning applies to the **npx stdio** transport only. A hosted HTTP entry
(`\{"type": "http", "url": "..."\}`) carries no npm package and therefore nothing to pin;
upstream versions it server-side. If you need a frozen version, you are choosing stdio.
To freeze a stdio server at a known-good version, edit `.mcp.json`:
```diff
"context7": {
"command": "npx",
- "args": ["-y", "@upstash/context7-mcp@latest"]
+ "args": ["-y", "@upstash/context7-mcp@4.0.2"]
}
```
`4.0.2` is npm latest as of 2026-08-11. Note that 2.x, 3.x, and 4.x expose the identical
two tools (`resolve-library-id`, `query-docs`), so a context7 major bump is not a
consumer-side breaking event; pin for reproducibility, not for API safety.
Document the pin reason in `mcp-version-matrix.md` under the affected row. Remove the pin once the reason is resolved (e.g., upstream ships a fix, your consumer code adapts to the new API).
## CC 2.1.128: MCP Reconnect Tool Summarization
CC 2.1.128 changed the on-reconnect tool announcement format. Re-announced tools are summarized by server prefix instead of being enumerated line-by-line:
```text
# Before (≤ 2.1.127)
mcp__github__create_issue re-registered
mcp__github__create_pull_request re-registered
mcp__github__list_repositories re-registered
... (37 lines per reconnect)
# After (≥ 2.1.128)
mcp__github__* (37 tools re-registered)
```
**Audit-log impact**: any audit-log diff/grep pattern that counted re-registrations by parsing per-line entries underestimates after 2.1.128. Two changes to apply:
1. **Source-of-truth for tool surface area** — use the **initial connect** event (first registration after server start), which still emits the full per-tool list. Treat reconnect events as deltas only.
2. **Reconnect-frequency monitoring** — the new summary line still contains the server prefix and tool count, so reconnect-event counters and per-server reconnect histograms continue to work; what changes is the per-tool dimension on reconnects.
**Grep recipes** for the audit log:
```bash
# Count reconnects per server (works on both formats)
grep -E '^mcp__([^_]+)__' audit.log | awk -F__ '{print $2}' | sort | uniq -c
# Tool inventory at session start (always full enumeration)
awk '/^=== session start ===/{in_session=1; next}
/^=== session end ===/{in_session=0}
in_session && /^mcp__/' audit.log
# Detect 2.1.128+ summary lines specifically
grep -E '^mcp__[^_]+__\* \([0-9]+ tools re-registered\)$' audit.log
```
If you maintain an external system that tracks tool counts via the audit log alone (e.g., a "MCP tool sprawl" alert), prefer the registry-discovery walk (`registry.modelcontextprotocol.io`) or query the server's `tools/list` directly via Inspector for an authoritative count.
## Related
- `mcp-version-matrix.md` — the current-state matrix this runbook maintains
- `.mcp.json` — the target of pin changes
- Issue #1462 — doctor check that will warn when HIGH-tier servers are unpinned
### Mcp Version Matrix
# MCP Version Matrix
Tracks OrchestKit's integrated MCP servers against upstream latest. Sourced from `.mcp.json` (project-level, user-owned) and documented user-level integrations.
**Last audited:** 2026-08-18 (context7 refresh: 2.2.0 was two majors stale; hosted HTTP recorded as the recommended transport)
## Audit Method
1. `.mcp.json` enumerates project-level servers
2. `npm view <pkg> version` + `time.modified` for each
3. Primary consumer located via `grep -rln "mcp__<name>__"` in `src/skills/` and `src/agents/`
## Matrix
| MCP server | Package | Pin in `.mcp.json` | Upstream latest | Last published | Δ since prior audit | Primary consumer | Status |
|---|---|---|---|---|---|---|---|
| **context7** | `@upstash/context7-mcp` (stdio) **or** hosted `https://mcp.context7.com/mcp` | `@latest`, hosted URL unversioned | **4.0.2** | 2026-08-11 | 2.2.0 → 4.0.2 (**two majors**) | 22 of 36 agents, plus chain-patterns, implement, fix-issue, cover | Active. Both majors expose the **same two tools**, so the bump grants nothing new |
| **sequential-thinking** | `@modelcontextprotocol/server-sequential-thinking` | unpinned | **2025.12.18** | 2026-02-06 | unchanged | chain-patterns, brainstorm, setup | Stable, calendar-versioned |
| **memory** | `@modelcontextprotocol/server-memory` | unpinned | **2026.1.26** | 2026-02-06 | unchanged | 46 files (cross-cutting) | Stable, calendar-versioned |
| **tavily** | `tavily-mcp` | `@latest` | **0.2.19** | 2026-04-24 | 0.2.18 → 0.2.19 (patch) | chain-patterns fallback (tier-fallbacks.md) | Low direct usage |
| **21st-dev-magic** | `@21st-dev/magic` | **`@0.1.0` (pinned)** | **0.1.0** | 2025-12-23 | unchanged | None in core (mentioned as option in `component-search`) | Pre-1.0, stale upstream — pinned 2026-04-26 to lock current state |
| **fal** | `fal-ai-mcp` | `@latest` | **0.2.1** | 2026-03-07 | unchanged | None in core (available for multimodal skills) | Active upstream |
## User-level MCPs (referenced but not in project `.mcp.json`)
| MCP server | Package | Referenced by | Note |
|---|---|---|---|
| **notebooklm-mcp** | `notebooklm-mcp-cli` (PyPI, **0.6.1**, 2026-04-28) — installed via `uv tool install notebooklm-mcp-cli`; ships both `nlm` CLI and `notebooklm-mcp` MCP server. NOTE: the npm `notebooklm-mcp@1.2.1` (2025-12-27) is a separate, stale package — do NOT use it. | `src/skills/release-sync/SKILL.md`, `src/skills/notebooklm/SKILL.md` | Configured user-level in `~/.claude.json`; `release-sync` assumes availability |
## Status: One HIGH-Tier Server Pinned, Rest on `@latest`
Of 6 active remote MCPs, 1 is now version-pinned (`@21st-dev/magic@0.1.0`) and 5 still resolve to `@latest`. The pinning of the HIGH-tier server addresses the original audit's primary risk: a breaking pre-1.0 upstream change can no longer propagate silently on the next `npx -y` fetch.
**Remaining MEDIUM-tier @latest entries** (`context7`, `tavily`, `fal`) are intentionally left on `@latest` — they're semver-disciplined upstream and benefit from automatic patch/minor uptake. The doctor check (`/ork:doctor` Category 12 sub-check) emits an informational note but does not warn, matching the tier policy.
## Transport: hosted HTTP is the recommendation for context7
context7 publishes a hosted server at `https://mcp.context7.com/mcp`, and that is the
transport to recommend:
```json
"context7": { "type": "http", "url": "https://mcp.context7.com/mcp" }
```
npx stdio remains supported and is the fallback for proxied or air-gapped networks, but
**it costs one child process per Claude Code session**, because the client spawns the
server. On a machine running ten concurrent sessions that is ten `npx` children resolving
and running the package, per project.
Two consequences for this matrix:
- A hosted entry has **no npm version to pin**. `check-mcp-pinning.sh` finds no package
token in it and classifies the server as `<unknown>`, which it skips. That is correct
behaviour, not a regression: version pinning is a property of the npx transport only.
- The version column above still matters for the stdio fallback. `4.0.2` is the concrete
pin to use when one is needed.
### Tool surface (unchanged across 2.x, 3.x, and 4.x)
| Tool | Purpose |
|---|---|
| `mcp__context7__resolve-library-id` | Map a package name to a Context7 library id |
| `mcp__context7__query-docs` | Fetch scoped documentation for a resolved library |
Exactly two, on every published version. Any other spelling grants nothing and fails
silently, so a version bump is never a reason to rewrite a tool name.
### Tier and key
Free is 1,000 requests against public repositories, with no key. Context7 Pro is $10 per
seat per month for 5,000 requests per seat plus private repo parsing, then $10 per
additional 1,000. Keys have the form `ctx7sk-...` and are supplied as
`Authorization: Bearer $\{CONTEXT7_API_KEY\}` on the hosted transport or `--api-key` on
stdio. Never write a literal key into a tracked file.
**Auth caveat.** `tools/list` returns HTTP 200 for every auth state (no header, garbage bearer,
unexpanded `$\{CONTEXT7_API_KEY\}` literal, valid key), so a connection check can never
confirm a key. Only `tools/call` discriminates: a bad key returns
`Invalid API key. Please check your API key. API keys should start with 'ctx7sk' prefix.`
**No anonymous fallback.** Hosted plus bearer is a hard dependency on `CONTEXT7_API_KEY`
being exported. With it unset, the unexpanded literal is transmitted as the token and every
`tools/call` fails; the server does not fall back to the free tier. Keyless hosted (no
`headers` block) is a working free tier, so an unexported key is worse than no header.
## Risk Tier (unchanged)
| Tier | Criteria | Servers |
|---|---|---|
| **LOW** | Stable API, calendar-versioned, low release velocity | `sequential-thinking`, `memory` |
| **MEDIUM** | Active upstream, semver, used in many skills | `context7`, `tavily`, `fal` |
| **HIGH** | Pre-1.0 upstream, API may change without notice | `21st-dev-magic` (0.1.0, **now pinned**) |
## Recommendations Status
| # | Action | Status | Notes |
|---|---|---|---|
| 1 | Pin HIGH-tier servers to concrete versions in `.mcp.json` | **DONE** (2026-04-26) | `21st-dev-magic` pinned to `@0.1.0` |
| 2 | Add an `ork:doctor` check that warns when `.mcp.json` uses `@latest` on HIGH-tier servers | **DONE** (M117 #1462) | Implemented in `src/skills/doctor/scripts/check-mcp-pinning.sh` (PR #1496) |
| 3 | Document in `ork:release-sync` that NotebookLM MCP is user-configured and not auto-installed | OPEN | Small skill edit, low priority |
| 4 | Re-run this audit every 90 days; update `Last audited` header | RECURRING | Next due: 2026-11-16 (last run 2026-08-18; the 2026-07-25 slot was missed, which is how context7 drifted two majors) |
## Audit Cadence Calibration
**Revised 2026-08-18: the 90-day cadence is only appropriate if it is actually run.**
The earlier version of this note argued the cadence was fine because `@latest` handles
version tracking continuously, leaving the audit to catch tier-reclassification signals
(a server going stale, hitting 1.0.0, or losing maintainer activity). Between audits 1
(2026-04-22) and 2 (2026-04-26) that held: two patch-level changes landed
(`context7 2.2.0`, `tavily 0.2.19`) and both were auto-uptaken with no user action.
What actually happened next falsifies the comfortable half of that argument. The
2026-07-25 slot was missed, and by 2026-08-18 this matrix still said context7's latest
was `2.2.0` while npm said `4.0.2`: **two major versions of documented drift**, roughly
four months stale. `@latest` did keep fetching the new code, so runtime was never wrong;
the *documentation* was, which is exactly the failure a version matrix exists to prevent.
It also hid a transport change worth knowing about (a hosted HTTP server now exists).
Keep the 90-day cadence, and treat a missed slot as a defect rather than a slip. Two
mitigations that cost nothing:
- Record the next due date in the Recommendations table (done) so a missed run is visible
in the file itself rather than only in someone's calendar.
- Audit the **transport** alongside the version. A major bump that changes no tool name is
low-risk; a new hosted endpoint that removes a per-session child process is not, and the
version column alone would never have surfaced it.
## How to Re-run This Audit
See `mcp-audit-runbook.md` (sibling reference) for the re-run script, interpretation rules, and when to escalate a drift. One-liner refresh:
```bash
for pkg in @upstash/context7-mcp tavily-mcp fal-ai-mcp @21st-dev/magic \
@modelcontextprotocol/server-sequential-thinking @modelcontextprotocol/server-memory \
notebooklm-mcp; do
v=$(npm view "$pkg" version 2>/dev/null)
pub=$(npm view "$pkg" time.modified 2>/dev/null)
printf "%-50s %-12s %s\n" "$pkg" "${v:-???}" "${pub:-unknown}"
done
```
## References
- `.mcp.json` — project-level MCP server config
- `~/.claude.json` — user-level MCP server config (not in repo)
- Issue #1446 — the audit request that produced this matrix
- PR #1496 (M117) — `/ork:doctor` MCP pinning sub-check
- `src/skills/mcp-patterns/SKILL.md` — skill that owns this reference
- `src/skills/doctor/references/mcp-pinning-check.md` — tier source-of-truth for the doctor check
- `mcp-audit-runbook.md` — the operational procedure
---
## Checklists (1)
### Mcp Server Checklist
# MCP Server Pre-Deployment Checklist
## Server Setup
- [ ] FastMCP lifespan used for resource management
- [ ] Transport selected (stdio for CLI, SSE for web, Streamable HTTP for production)
- [ ] All tools have descriptive docstrings
- [ ] Input validation on all tool arguments
- [ ] Error responses return text content (not exceptions)
## Security Hardening
- [ ] Zero-trust tool allowlist configured
- [ ] Tool description sanitization enabled
- [ ] Hash verification on tool invocations
- [ ] No secrets in tool output (API keys, credentials)
- [ ] Human-in-the-loop for high-risk operations
- [ ] Encoding normalization applied to inputs
- [ ] Injection pattern detection active
## Resource Management
- [ ] Response sizes bounded (Claude context limits)
- [ ] Resource caching with TTL and LRU eviction
- [ ] Memory cap configured for resource cache
- [ ] No blocking synchronous code in async server
## Testing
- [ ] Tool invocations tested with valid and invalid inputs
- [ ] Security rules validated against injection payloads
- [ ] Transport failover tested (reconnect behavior)
- [ ] Lifecycle cleanup verified (no leaked connections)