mcp-advanced-patterns（MCP 高级模式）

MCP Advanced Patterns

Advanced Model Context Protocol patterns for production-grade MCP implementations.

FastMCP 2.14.x (Jan): Enterprise auth, OpenAPI/FastAPI generation, server composition, proxying. Python 3.10-3.13.

Overview

• Composing multiple tools into orchestrated workflows
• Managing resource lifecycle and caching efficiently
• Scaling MCP servers horizontally with load balancing
• Building custom MCP servers with middleware and transports
• Implementing auto-enable thresholds for context management

Tool Composition Pattern

from dataclasses import dataclass
from typing import Any, Callable, Awaitable

@dataclass
class ComposedTool:
    """Combine multiple tools into a single pipeline operation."""
    name: str
    tools: dict[str, Callable[..., Awaitable[Any]]]
    pipeline: list[str]

    async def execute(self, input_data: dict[str, Any]) -> dict[str, Any]:
        """Execute tool pipeline sequentially."""
        result = input_data
        for tool_name in self.pipeline:
            tool = self.tools[tool_name]
            result = await tool(result)
        return result

# Example: Search + Summarize composition
search_summarize = ComposedTool(
    name="search_and_summarize",
    tools={
        "search": search_documents,
        "summarize": summarize_content,
    },
    pipeline=["search", "summarize"]
)

FastMCP Server with Lifecycle

from contextlib import asynccontextmanager
from collections.abc import AsyncIterator
from dataclasses import dataclass
from mcp.server.fastmcp import Context, FastMCP

@dataclass
class AppContext:
    """Typed application context with shared resources."""
    db: Database
    cache: CacheService
    config: dict

@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
    """Manage server startup and shutdown lifecycle."""
    # Initialize on startup
    db = await Database.connect()
    cache = await CacheService.connect()

    try:
        yield AppContext(db=db, cache=cache, config={"timeout": 30})
    finally:
        # Cleanup on shutdown
        await cache.disconnect()
        await db.disconnect()

mcp = FastMCP("Production Server", lifespan=app_lifespan)

@mcp.tool()
def query_data(sql: str, ctx: Context) -> str:
    """Execute query using shared connection."""
    app_ctx = ctx.request_context.lifespan_context
    return app_ctx.db.query(sql)

Auto-Enable Thresholds (CC 2.1.9)

Configure MCP servers to auto-enable/disable based on context window usage:

# .claude/settings.json
mcp:
  context7:
    enabled: auto:75    # High-value docs, keep available longer
  sequential-thinking:
    enabled: auto:60    # Complex reasoning needs room
  memory:
    enabled: auto:90    # Knowledge graph - preserve until compaction
  playwright:
    enabled: auto:50    # Browser-heavy, disable early

Threshold Guidelines:

Resource Management

from functools import lru_cache
from datetime import datetime, timedelta
from typing import Any

class MCPResourceManager:
    """Manage MCP resources with caching and lifecycle."""

    def __init__(self, cache_ttl: timedelta = timedelta(minutes=15)):
        self.resources: dict[str, Any] = {}
        self.cache_ttl = cache_ttl
        self.last_access: dict[str, datetime] = {}

    def get_resource(self, uri: str) -> Any:
        """Get resource with access time tracking."""
        if uri in self.resources:
            self.last_access[uri] = datetime.now()
            return self.resources[uri]

        resource = self._load_resource(uri)
        self.resources[uri] = resource
        self.last_access[uri] = datetime.now()
        return resource

    def cleanup_stale(self) -> int:
        """Remove stale resources. Returns count of removed."""
        now = datetime.now()
        stale = [
            uri for uri, last in self.last_access.items()
            if now - last > self.cache_ttl
        ]
        for uri in stale:
            del self.resources[uri]
            del self.last_access[uri]
        return len(stale)

Horizontal Scaling

import asyncio
from typing import List

class MCPLoadBalancer:
    """Load balance across multiple MCP server instances."""

    def __init__(self, servers: List[str]):
        self.servers = servers
        self.current = 0
        self.health: dict[str, bool] = {s: True for s in servers}

    async def get_healthy_server(self) -> str:
        """Round-robin with health check."""
        for _ in range(len(self.servers)):
            server = self.servers[self.current]
            self.current = (self.current + 1) % len(self.servers)
            if self.health[server]:
                return server
        raise RuntimeError("No healthy servers available")

    async def health_check_loop(self):
        """Periodic health check for all servers."""
        while True:
            for server in self.servers:
                try:
                    self.health[server] = await self._ping(server)
                except Exception:
                    self.health[server] = False
            await asyncio.sleep(30)

Key Decisions

Common Mistakes

• No lifecycle management (resource leaks)
• Missing health checks in load balancing
• Hardcoded server endpoints
• No graceful degradation on server failure
• Ignoring context window thresholds

Related Skills

• function-calling - LLM tool integration patterns
• resilience-patterns - Circuit breakers and retries
• connection-pooling - Database connection management
• streaming-api-patterns - Real-time streaming

Capability Details

tool-composition

Keywords: tool composition, pipeline, orchestration, chain tools Solves:

• Combine multiple tools into workflows
• Sequential tool execution
• Tool result passing

resource-management

Keywords: resource, cache, lifecycle, cleanup, ttl Solves:

• Manage resource lifecycle
• Implement resource caching
• Clean up stale resources

scaling-strategies

Keywords: scale, load balance, horizontal, health check, redundancy Solves:

• Scale MCP servers horizontally
• Implement health-checked load balancing
• Handle server failures gracefully

server-building

Keywords: server, fastmcp, lifespan, middleware, transport Solves:

• Build production MCP servers
• Manage server lifecycle
• Configure transports and middleware

auto-enable-thresholds

Keywords: auto-enable, context window, threshold, auto:N Solves:

• Configure MCP auto-enable/disable
• Manage context window usage
• Optimize MCP server availability