MCP Patterns

search_kb
  WHEN NOT TO USE: do not use to fetch a document you already have the id for —
    that is get_document. Use search_kb only when you need to discover *which*
    document, by meaning or keyword.

get_document
  WHEN NOT TO USE: do not use to find a document by topic or keyword — you must
    already hold an exact id (from search_kb results). For discovery, use search_kb.

name: cancel_workflow
description: |
  Stop a running agent workflow and discard its in-flight results.

  WHEN TO USE: the user says "cancel the workflow", "stop run abc123",
    "kill that job", or asks to halt a workflow that get_workflow_status
    reports as RUNNING.

  WHEN NOT TO USE:
    - To inspect progress without stopping — use get_workflow_status.
    - To start a fresh run — use start_workflow.
    - On a workflow already in a terminal state (COMPLETED/FAILED) — the call
      is a no-op and signals the model misread the status.

  CRITICAL: cancellation is irreversible and drops partial output. Confirm the
    workflowId came from list_workflows or get_workflow_status — never type one
    from memory.

  Self-test: am I stopping work that is genuinely still RUNNING, or did I confuse
  "check status" with "cancel"? If I have not seen a RUNNING status, do not call this.

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

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

# 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)

@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))

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

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

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

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 } };
});

// 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>();

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)