mcp-patterns

name	mcp-patterns
description	MCP server design: tool schemas, resources, stdio/SSE, capability negotiation. Triggers: MCP, Model Context Protocol, JSON-RPC, stdio, SSE, Claude Desktop.
effort	medium
user-invocable	false
allowed-tools	Read

MCP Patterns Skill

MCP Specification (2025-06-18)

Core Concepts

Concept	Description
Server	Exposes tools, resources, prompts to clients
Client	Connects to servers, invokes tools
Transport	Communication layer (stdio, HTTP, SSE)
Tool	Executable function with JSON Schema
Resource	Read-only data (files, URLs)
Prompt	Reusable prompt template

---

Tool Definition Pattern

// TypeScript with @modelcontextprotocol/sdk
server.setRequestHandler(ListToolsRequestSchema, async () => ({
  tools: [{
    name: "search_kb",
    description: "Search the knowledge base",
    inputSchema: {
      type: "object",
      properties: {
        query: {
          type: "string",
          description: "Search query"
        },
        limit: {
          type: "number",
          description: "Max results",
          default: 10
        }
      },
      required: ["query"]
    },
    annotations: {
      readOnlyHint: true,      // Doesn't modify state
      idempotentHint: true,    // Same input = same output
      openWorldHint: false     // Bounded result set
    }
  }]
}));

Tool Annotations

Annotation	Meaning	Use When
readOnlyHint	No side effects	Read operations
destructiveHint	Deletes/modifies data	Write operations
idempotentHint	Safe to retry	GET-like operations
openWorldHint	Results may change	External API calls

---

Transport Patterns

1. stdio (Default)

import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio";

const transport = new StdioServerTransport();
await server.connect(transport);

2. HTTP with SSE

# FastAPI implementation
from fastapi import FastAPI
from sse_starlette.sse import EventSourceResponse

app = FastAPI()

@app.get("/mcp/sse")
async def sse_endpoint():
    async def event_generator():
        while True:
            # Yield MCP events
            yield {"event": "message", "data": json.dumps(response)}
    return EventSourceResponse(event_generator())

@app.post("/mcp")
async def jsonrpc_endpoint(request: Request):
    body = await request.json()
    response = await handle_jsonrpc(body)
    return JSONResponse(response)

3. Streamable HTTP (Modern)

Single /mcp endpoint handling GET (SSE) and POST (JSON-RPC):

@app.api_route("/mcp", methods=["GET", "POST"])
async def mcp_endpoint(request: Request):
    if request.method == "GET":
        # SSE streaming
        return EventSourceResponse(stream_generator())
    else:
        # JSON-RPC
        body = await request.json()
        return JSONResponse(await handle_jsonrpc(body))

---

JSON-RPC 2.0 Pattern

Request Format

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "search_kb",
    "arguments": {"query": "test", "limit": 5}
  }
}

Response Format

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "content": [
      {"type": "text", "text": "Search results..."}
    ]
  }
}

Error Format

{
  "jsonrpc": "2.0",
  "id": 1,
  "error": {
    "code": -32601,
    "message": "Method not found"
  }
}

Standard Error Codes

Code	Meaning
-32700	Parse error
-32600	Invalid request
-32601	Method not found
-32602	Invalid params
-32603	Internal error

---

Completion Support

Enable intelligent argument suggestions:

server.setRequestHandler(CompletionCompleteRequestSchema, async (request) => {
  const { argument } = request.params;

  if (argument.name === "service") {
    return {
      completion: {
        values: ["nginx", "postgresql", "redis", "qdrant"],
        hasMore: false
      }
    };
  }

  return { completion: { values: [], hasMore: false } };
});

---

Session Management

// Generate secure session ID
const sessionId = crypto.randomUUID();

// Validate Origin header
const allowedOrigins = ["http://localhost:3000", "https://claude.ai"];
if (!allowedOrigins.includes(request.headers.origin)) {
  throw new Error("Invalid origin");
}

// Session storage (don't expose to client)
const sessions = new Map<string, SessionData>();

---

Batching Support

Handle multiple requests in single HTTP call:

async def handle_jsonrpc(body):
    if isinstance(body, list):
        # Batch request
        return [await process_single(req) for req in body]
    else:
        # Single request
        return await process_single(body)

---

Best Practices

Security

• Validate all inputs against JSON Schema
• Validate Origin header
• Implement rate limiting
• Use environment variables for secrets
• No secrets in logs or errors

Performance

• Connection pooling
• Response caching where appropriate
• Streaming for large responses
• Batch support for efficiency

Logging

• Log to stderr (never stdout)
• Structured JSON logs
• Request/response tracing
• Error logging with context

Documentation

• Document all tools clearly
• Include example usage
• Version your API
• Maintain changelog

---

RAG-MCP MCP Tools

Tool	Description
smart_query	Primary search with auto-routing
hybrid_search_kb	Raw vector + text search
get_document	Full document content
crag_search	Self-correcting search
multi_hop_search	Complex reasoning
start_workflow	Start agent workflow
get_workflow_status	Check workflow progress
list_workflows	List all workflows
cancel_workflow	Cancel workflow