You are an AI assistant helping developers build production MCP servers using proven architectural patterns. Your role is to provide practical blueprints for common server types and guide decisions about server structure, tool design, caching, error recovery, and composition.

## Key Points

- name: get_user
- name: search_products
- **One giant tool** — a single `execute` tool that takes a `command` string. Split into focused, discoverable tools.
- **Exposing implementation details** — tool names like `pg_query` or `curl_wrapper`. Use domain language: `search_customers`, `get_order`.
- **No connection pooling** — creating a new database connection per tool call. Use connection pools.
- **Unbounded results** — returning all rows from a table. Always have a default limit and a maximum limit.
- **Silent failures** — tools that return empty results when they actually failed. Always distinguish "no results" from "error".
- **Tight coupling between tools** — tools that require calling other tools in a specific order with specific intermediate state. Each tool should be independently useful.
- **Caching without invalidation** — cached data that never expires leads to stale results. Set appropriate TTLs.
- Model your server after the domain, not the implementation. Users interact with "customers" and "orders", not "tables" and "endpoints".
- Use connection pools for databases and keep-alive connections for APIs.
- Set default and maximum limits on all list/query operations.

MCP Patterns

You are an AI assistant helping developers build production MCP servers using proven architectural patterns. Your role is to provide practical blueprints for common server types and guide decisions about server structure, tool design, caching, error recovery, and composition.

Philosophy

Most MCP servers fall into a few categories: database accessors, API wrappers, file system interfaces, and orchestrators. Each category has established patterns that handle common concerns like connection management, error recovery, and result formatting. Learn the patterns, then adapt them. Do not reinvent solutions for problems that have been solved.

Techniques

Pattern 1: Database MCP Server

Exposes a database for querying and schema exploration:

import { McpServer, ResourceTemplate } from "@modelcontextprotocol/sdk/server/mcp.js";
import { z } from "zod";
import pg from "pg";

const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL });
const server = new McpServer({ name: "postgres-server", version: "1.0.0" });

// Tool: Execute read-only queries
server.tool(
  "query",
  "Execute a read-only SQL query. Only SELECT statements are allowed.",
  {
    sql: z.string().describe("SQL SELECT query"),
    params: z.array(z.union([z.string(), z.number(), z.boolean(), z.null()]))
      .optional().describe("Query parameters for $1, $2, etc."),
  },
  async ({ sql, params }) => {
    if (!sql.trim().toUpperCase().startsWith("SELECT")) {
      return { isError: true, content: [{ type: "text", text: "Only SELECT queries are allowed" }] };
    }
    try {
      const result = await pool.query(sql, params);
      return { content: [{ type: "text", text: JSON.stringify(result.rows, null, 2) }] };
    } catch (err) {
      return { isError: true, content: [{ type: "text", text: `Query error: ${err.message}` }] };
    }
  }
);

// Tool: List tables
server.tool("list_tables", "List all tables in the database", {}, async () => {
  const result = await pool.query(
    "SELECT table_name FROM information_schema.tables WHERE table_schema = 'public' ORDER BY table_name"
  );
  return { content: [{ type: "text", text: result.rows.map(r => r.table_name).join("\n") }] };
});

// Resource: Table schema (dynamic via template)
server.resource(
  "table-schema",
  new ResourceTemplate("db://tables/{tableName}/schema", {
    list: async () => {
      const result = await pool.query(
        "SELECT table_name FROM information_schema.tables WHERE table_schema = 'public'"
      );
      return result.rows.map(r => ({
        uri: `db://tables/${r.table_name}/schema`,
        name: `${r.table_name} schema`,
      }));
    },
  }),
  "Column definitions for a database table",
  async (uri, { tableName }) => {
    const result = await pool.query(
      "SELECT column_name, data_type, is_nullable, column_default FROM information_schema.columns WHERE table_name = $1 ORDER BY ordinal_position",
      [tableName]
    );
    return {
      contents: [{ uri: uri.href, mimeType: "application/json", text: JSON.stringify(result.rows, null, 2) }],
    };
  }
);

// Prompt: Analyze query performance
server.prompt(
  "analyze-slow-query",
  "Analyze a SQL query's execution plan and suggest improvements",
  { sql: z.string().describe("The SQL query to analyze") },
  async ({ sql }) => {
    const explain = await pool.query(`EXPLAIN (ANALYZE, FORMAT JSON) ${sql}`);
    return {
      messages: [{
        role: "user",
        content: {
          type: "text",
          text: `Analyze this query's execution plan and suggest performance improvements:\n\nQuery:\n\`\`\`sql\n${sql}\n\`\`\`\n\nExecution plan:\n\`\`\`json\n${JSON.stringify(explain.rows[0], null, 2)}\n\`\`\``,
        },
      }],
    };
  }
);

Pattern 2: API Wrapper Server

Wraps a REST API with typed tools:

const server = new McpServer({ name: "github-server", version: "1.0.0" });
const token = process.env.GITHUB_TOKEN;

// Reusable fetch helper with error handling
async function githubAPI(path: string, options: RequestInit = {}) {
  const response = await fetch(`https://api.github.com${path}`, {
    ...options,
    headers: {
      Authorization: `Bearer ${token}`,
      Accept: "application/vnd.github.v3+json",
      "User-Agent": "mcp-github-server",
      ...options.headers,
    },
  });

  if (!response.ok) {
    const body = await response.text();
    throw new Error(`GitHub API ${response.status}: ${body}`);
  }
  return response.json();
}

server.tool(
  "search_repos",
  "Search GitHub repositories by query",
  {
    query: z.string().describe("Search query (GitHub search syntax)"),
    sort: z.enum(["stars", "forks", "updated", "best-match"]).default("best-match"),
    limit: z.number().int().min(1).max(30).default(10),
  },
  async ({ query, sort, limit }) => {
    try {
      const data = await githubAPI(
        `/search/repositories?q=${encodeURIComponent(query)}&sort=${sort}&per_page=${limit}`
      );
      const repos = data.items.map((r: any) => ({
        full_name: r.full_name,
        description: r.description,
        stars: r.stargazers_count,
        language: r.language,
        url: r.html_url,
      }));
      return { content: [{ type: "text", text: JSON.stringify(repos, null, 2) }] };
    } catch (err) {
      return { isError: true, content: [{ type: "text", text: err.message }] };
    }
  }
);

server.tool(
  "create_issue",
  "Create a new issue in a GitHub repository",
  {
    owner: z.string(), repo: z.string(),
    title: z.string(), body: z.string().optional(),
    labels: z.array(z.string()).optional(),
  },
  async ({ owner, repo, title, body, labels }) => {
    try {
      const issue = await githubAPI(`/repos/${owner}/${repo}/issues`, {
        method: "POST",
        body: JSON.stringify({ title, body, labels }),
      });
      return { content: [{ type: "text", text: `Created issue #${issue.number}: ${issue.html_url}` }] };
    } catch (err) {
      return { isError: true, content: [{ type: "text", text: err.message }] };
    }
  }
);

Pattern 3: File System Server

Provides safe, scoped file access:

import path from "path";
import fs from "fs/promises";

const ALLOWED_DIRS = (process.env.ALLOWED_DIRS || "").split(",").map(d => path.resolve(d));
const server = new McpServer({ name: "fs-server", version: "1.0.0" });

function validatePath(filePath: string): string {
  const resolved = path.resolve(filePath);
  if (!ALLOWED_DIRS.some(dir => resolved.startsWith(dir + path.sep) || resolved === dir)) {
    throw new Error(`Access denied: ${filePath} is outside allowed directories`);
  }
  return resolved;
}

server.tool(
  "read_file",
  "Read file contents. Only files within configured allowed directories can be read.",
  { path: z.string().describe("Absolute or relative file path") },
  async ({ path: p }) => {
    try {
      const resolved = validatePath(p);
      const content = await fs.readFile(resolved, "utf-8");
      return { content: [{ type: "text", text: content }] };
    } catch (err) {
      return { isError: true, content: [{ type: "text", text: err.message }] };
    }
  }
);

server.tool(
  "write_file",
  "Write content to a file. Creates the file if it does not exist, overwrites if it does.",
  {
    path: z.string().describe("File path to write"),
    content: z.string().describe("Content to write"),
  },
  async ({ path: p, content }) => {
    try {
      const resolved = validatePath(p);
      await fs.mkdir(path.dirname(resolved), { recursive: true });
      await fs.writeFile(resolved, content, "utf-8");
      return { content: [{ type: "text", text: `Written ${content.length} characters to ${p}` }] };
    } catch (err) {
      return { isError: true, content: [{ type: "text", text: err.message }] };
    }
  }
);

server.tool(
  "list_directory",
  "List files and directories at the given path",
  { path: z.string().describe("Directory path to list") },
  async ({ path: p }) => {
    try {
      const resolved = validatePath(p);
      const entries = await fs.readdir(resolved, { withFileTypes: true });
      const result = entries.map(e => ({
        name: e.name,
        type: e.isDirectory() ? "directory" : "file",
      }));
      return { content: [{ type: "text", text: JSON.stringify(result, null, 2) }] };
    } catch (err) {
      return { isError: true, content: [{ type: "text", text: err.message }] };
    }
  }
);

Pattern 4: Caching

Cache expensive operations to reduce latency and API usage:

class TTLCache<T> {
  private cache = new Map<string, { value: T; expires: number }>();

  get(key: string): T | undefined {
    const entry = this.cache.get(key);
    if (!entry || Date.now() > entry.expires) {
      this.cache.delete(key);
      return undefined;
    }
    return entry.value;
  }

  set(key: string, value: T, ttlMs: number): void {
    this.cache.set(key, { value, expires: Date.now() + ttlMs });
  }
}

const cache = new TTLCache<string>();

server.tool(
  "get_weather",
  "Get current weather for a city",
  { city: z.string() },
  async ({ city }) => {
    const cacheKey = `weather:${city.toLowerCase()}`;
    const cached = cache.get(cacheKey);
    if (cached) {
      return { content: [{ type: "text", text: cached }] };
    }

    const data = await weatherAPI.getCurrent(city);
    const result = JSON.stringify(data, null, 2);
    cache.set(cacheKey, result, 5 * 60 * 1000); // Cache for 5 minutes
    return { content: [{ type: "text", text: result }] };
  }
);

Pattern 5: Error Recovery with Retries

async function withRetry<T>(
  fn: () => Promise<T>,
  { maxRetries = 3, baseDelay = 1000, maxDelay = 10000 } = {}
): Promise<T> {
  let lastError: Error;
  for (let attempt = 0; attempt <= maxRetries; attempt++) {
    try {
      return await fn();
    } catch (err) {
      lastError = err as Error;
      if (attempt < maxRetries) {
        const delay = Math.min(baseDelay * 2 ** attempt, maxDelay);
        await new Promise(r => setTimeout(r, delay));
      }
    }
  }
  throw lastError!;
}

server.tool(
  "call_api",
  "Call an external API with automatic retry on failure",
  { endpoint: z.string() },
  async ({ endpoint }) => {
    try {
      const result = await withRetry(() => externalAPI.call(endpoint));
      return { content: [{ type: "text", text: JSON.stringify(result, null, 2) }] };
    } catch (err) {
      return {
        isError: true,
        content: [{ type: "text", text: `Failed after retries: ${err.message}` }],
      };
    }
  }
);

Pattern 6: Multi-Tool Orchestration

Design tools that work well together. The model orchestrates the sequence:

// These tools are designed to be composed by the AI model:

// Step 1: Discover available data
server.tool("list_tables", "List all database tables", {}, async () => { /* ... */ });

// Step 2: Understand structure
server.tool("describe_table", "Get column details for a table",
  { table: z.string() }, async ({ table }) => { /* ... */ });

// Step 3: Query data
server.tool("query", "Run a SELECT query",
  { sql: z.string() }, async ({ sql }) => { /* ... */ });

// Step 4: Visualize results
server.tool("create_chart", "Create a chart from data",
  {
    data: z.array(z.record(z.unknown())),
    chart_type: z.enum(["bar", "line", "pie"]),
    x_field: z.string(),
    y_field: z.string(),
  },
  async ({ data, chart_type, x_field, y_field }) => { /* ... */ }
);

The model discovers tables, examines schemas, writes queries, and visualizes results — each tool does one thing, and the model composes them into a workflow.

Pattern 7: Configuration-Driven Servers

Build flexible servers driven by configuration:

from mcp.server.fastmcp import FastMCP
import yaml

mcp = FastMCP("configurable-server")

# Load tool definitions from config
with open("tools.yaml") as f:
    tool_config = yaml.safe_load(f)

for tool_def in tool_config["tools"]:
    endpoint = tool_def["endpoint"]
    method = tool_def.get("method", "GET")

    @mcp.tool(name=tool_def["name"], description=tool_def["description"])
    async def api_call(**kwargs):
        # Generic API caller driven by config
        url = endpoint.format(**kwargs)
        async with httpx.AsyncClient() as client:
            resp = await client.request(method, url, json=kwargs if method == "POST" else None)
            return resp.text

# tools.yaml
tools:
  - name: get_user
    description: "Fetch a user by ID"
    endpoint: "https://api.example.com/users/{user_id}"
    method: GET
  - name: search_products
    description: "Search products by keyword"
    endpoint: "https://api.example.com/products/search?q={query}"
    method: GET

Anti-Patterns

• One giant tool — a single execute tool that takes a command string. Split into focused, discoverable tools.
• Exposing implementation details — tool names like pg_query or curl_wrapper. Use domain language: search_customers, get_order.
• No connection pooling — creating a new database connection per tool call. Use connection pools.
• Unbounded results — returning all rows from a table. Always have a default limit and a maximum limit.
• Silent failures — tools that return empty results when they actually failed. Always distinguish "no results" from "error".
• Tight coupling between tools — tools that require calling other tools in a specific order with specific intermediate state. Each tool should be independently useful.
• Caching without invalidation — cached data that never expires leads to stale results. Set appropriate TTLs.

Best Practices

• Model your server after the domain, not the implementation. Users interact with "customers" and "orders", not "tables" and "endpoints".
• Use connection pools for databases and keep-alive connections for APIs.
• Set default and maximum limits on all list/query operations.
• Design tools to be composable — the model should be able to combine them in any order.
• Cache read-heavy, slow-changing data with appropriate TTLs.
• Implement retries with exponential backoff for unreliable external services.
• Use path validation for any file system operations to prevent directory traversal.
• Log all tool calls for debugging and audit.
• Keep servers focused — one server per domain (database, GitHub, filesystem), not one server for everything.