Dec 23, 2025

FastMCP 深度解析：Python MCP 框架的架構設計與實戰指南

0. 前言：什麼是 FastMCP？

FastMCP 是一個高階 Python 框架，用於快速建構 MCP（Model Context Protocol） 伺服器與客戶端。MCP 是 Anthropic 提出的開放協議，目的是標準化 LLM 應用程式與外部工具、資源、提示詞之間的通訊方式。

FastMCP 的設計理念類似於 FastAPI 之於 REST API——透過 Python 裝飾器和型別註解，讓開發者以最少的程式碼建構功能完整的 MCP 服務。

核心特色

裝飾器驅動：使用 @mcp.tool、@mcp.resource、@mcp.prompt 快速定義組件
自動 Schema 生成：基於 Pydantic 和 Python 型別註解自動產生 JSON Schema
依賴注入：類似 FastAPI 的 Depends() 模式，支援 Context、Docket 等注入
中間件系統：可插拔的請求攔截與處理管線
伺服器組合：透過 mount() 和 Provider 實現模組化架構
完整認證：內建 OAuth 2.0/OIDC 支援，整合 GitHub、Google、Azure 等提供者
背景任務：支援 SEP-1686 任務協議，可執行長時間運算

1. 項目結構總覽

1.1 目錄架構

fastmcp/
├── src/fastmcp/
│   ├── __init__.py              # 公開 API 入口
│   ├── settings.py              # 全局設置 (Pydantic Settings)
│   ├── exceptions.py            # 自定義異常類
│   ├── mcp_config.py            # MCP 配置解析
│   │
│   ├── server/                  # 伺服器核心
│   │   ├── server.py            # FastMCP 主類
│   │   ├── context.py           # Context 上下文機制
│   │   ├── dependencies.py      # 依賴注入系統
│   │   ├── http.py              # HTTP/SSE 傳輸層
│   │   ├── middleware/          # 中間件系統
│   │   ├── providers/           # 動態組件提供者
│   │   ├── auth/                # 認證系統
│   │   │   ├── auth.py          # AuthProvider 基類
│   │   │   └── providers/       # OAuth 提供者實現
│   │   └── tasks/               # 背景任務 (SEP-1686)
│   │
│   ├── client/                  # 客戶端實現
│   │   ├── client.py            # Client 主類
│   │   ├── transports.py        # 傳輸層實現
│   │   └── auth/                # 客戶端認證
│   │
│   ├── tools/                   # 工具系統
│   │   ├── tool.py              # Tool, FunctionTool
│   │   └── tool_manager.py      # ToolManager
│   │
│   ├── resources/               # 資源系統
│   │   ├── resource.py          # Resource, FunctionResource
│   │   └── template.py          # ResourceTemplate
│   │
│   └── prompts/                 # 提示系統
│       └── prompt.py            # Prompt, FunctionPrompt

1.2 類別繼承關係

classDiagram
    class FastMCPBaseModel {
        +model_config: ConfigDict
    }

    class FastMCPComponent {
        +name: str
        +description: str
        +tags: set[str]
        +enabled: bool
        +enable()
        +disable()
    }

    class Tool {
        +parameters: dict
        +output_schema: dict
        +run(arguments) ToolResult
        +to_mcp_tool() MCPTool
    }

    class FunctionTool {
        +fn: Callable
        +from_function() FunctionTool
    }

    class Resource {
        +uri: AnyUrl
        +mime_type: str
        +read() ResourceContent
    }

    class Prompt {
        +arguments: list[PromptArgument]
        +render(arguments) PromptResult
    }

    class Provider {
        +list_tools() Sequence[Tool]
        +get_tool(name) Tool
        +list_resources() Sequence[Resource]
    }

    FastMCPBaseModel <|-- FastMCPComponent
    FastMCPComponent <|-- Tool
    FastMCPComponent <|-- Resource
    FastMCPComponent <|-- Prompt
    Tool <|-- FunctionTool
    Resource <|-- FunctionResource
    Prompt <|-- FunctionPrompt
    Provider <|-- FastMCPProvider
    Provider <|-- ProxyProvider
    Provider <|-- TransformingProvider

2. 核心架構解析

2.1 FastMCP 主類

FastMCP 是框架的核心入口，負責管理所有 MCP 組件：

from fastmcp import FastMCP

# 建立伺服器實例
mcp = FastMCP(
    name="My Server",
    instructions="這是一個範例伺服器",
    auth=None,  # 或 AuthProvider 實例
    middleware=[],  # 中間件列表
    providers=[],  # 動態組件提供者
)

內部結構：

class FastMCP(Generic[LifespanResultT]):
    def __init__(
        self,
        name: str | None = None,
        instructions: str | None = None,
        *,
        auth: AuthProvider | None = None,
        middleware: Sequence[Middleware] | None = None,
        providers: Sequence[Provider] | None = None,
        lifespan: LifespanCallable | None = None,
        mask_error_details: bool | None = None,
        tools: Sequence[Tool | Callable[..., Any]] | None = None,
        # ... 更多參數
    ):
        # 管理器初始化
        self._tool_manager: ToolManager
        self._resource_manager: ResourceManager
        self._prompt_manager: PromptManager
        self._providers: list[Provider]

        # Docket (背景任務) 相關
        self._docket: Docket | None = None

        # 底層 MCP Server
        self._mcp_server: LowLevelServer

2.2 請求處理流程

sequenceDiagram
    participant Client as MCP Client
    participant Transport as Transport Layer
    participant MW as Middleware Chain
    participant Server as FastMCP Server
    participant Handler as Request Handler
    participant Component as Tool/Resource/Prompt

    Client->>Transport: MCP Request
    Transport->>MW: MiddlewareContext

    loop Each Middleware
        MW->>MW: on_message()
        MW->>MW: on_request()
        MW->>MW: on_call_tool() / on_read_resource()
    end

    MW->>Handler: Processed Request
    Handler->>Server: get_tool() / get_resource()
    Server->>Component: run() / read() / render()
    Component-->>Server: Result
    Server-->>Handler: Formatted Response
    Handler-->>MW: Response
    MW-->>Transport: Final Response
    Transport-->>Client: MCP Response

3. 裝飾器系統深度剖析

3.1 Tool 裝飾器

FastMCP 的裝飾器支援多種呼叫模式：

from fastmcp import FastMCP

mcp = FastMCP("Demo")

# 模式 1：無參數裝飾器
@mcp.tool
def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b

# 模式 2：帶參數裝飾器
@mcp.tool(name="multiply", description="Multiply two numbers")
def mul(a: int, b: int) -> int:
    return a * b

# 模式 3：帶標籤和 metadata
@mcp.tool(tags={"math", "basic"}, meta={"version": "1.0"})
def subtract(a: int, b: int) -> int:
    """Subtract b from a"""
    return a - b

裝飾器內部實現：

def tool(
    self,
    name_or_fn: str | AnyFunction | None = None,
    *,
    name: str | None = None,
    description: str | None = None,
    tags: set[str] | None = None,
    # ...更多參數
) -> Callable[[AnyFunction], FunctionTool] | FunctionTool:
    """支持多種調用模式的裝飾器"""

    # 檢測是否直接傳入函數 (@tool 無括號形式)
    if inspect.isroutine(name_or_fn):
        fn = name_or_fn
        tool = Tool.from_function(fn, name=name, ...)
        self.add_tool(tool)
        return tool

    # 返回包裝函數用於帶參數調用 (@tool("name"))
    return partial(self.tool, name=tool_name, ...)

3.2 自動 JSON Schema 生成

FastMCP 利用 Pydantic 和 Python 型別註解自動生成 JSON Schema：

from typing import Annotated
from pydantic import BaseModel, Field

class ShrimpTank(BaseModel):
    class Shrimp(BaseModel):
        name: Annotated[str, Field(max_length=10)]

    shrimp: list[Shrimp]

@mcp.tool
def name_shrimp(
    tank: ShrimpTank,
    extra_names: Annotated[list[str], Field(max_length=10)],
) -> list[str]:
    """List all shrimp names in the tank"""
    return [shrimp.name for shrimp in tank.shrimp] + extra_names

生成的 JSON Schema：

{
  "type": "object",
  "properties": {
    "tank": {
      "type": "object",
      "properties": {
        "shrimp": {
          "type": "array",
          "items": {
            "type": "object",
            "properties": {
              "name": {"type": "string", "maxLength": 10}
            },
            "required": ["name"]
          }
        }
      },
      "required": ["shrimp"]
    },
    "extra_names": {
      "type": "array",
      "items": {"type": "string"},
      "maxLength": 10
    }
  },
  "required": ["tank", "extra_names"]
}

3.3 Resource 與 ResourceTemplate

# 靜態 Resource
@mcp.resource("config://settings")
def get_settings() -> dict:
    """Return application settings"""
    return {"theme": "dark", "language": "zh-TW"}

# 動態 Resource (URI 模板)
@mcp.resource("user://{user_id}/profile")
def get_user_profile(user_id: str) -> dict:
    """Get user profile by ID"""
    return {"id": user_id, "name": f"User {user_id}"}

# 帶 MIME 類型的 Resource
@mcp.resource("report://daily", mime_type="application/json")
def daily_report() -> dict:
    return {"date": "2025-01-01", "metrics": {...}}

3.4 Prompt 定義

from fastmcp.prompts.prompt import Message

@mcp.prompt("code_review")
def code_review_prompt(
    language: str,
    code: str,
    focus_areas: list[str] | None = None
) -> list[Message]:
    """Generate a code review prompt"""
    areas = ", ".join(focus_areas) if focus_areas else "general quality"

    return [
        Message(
            role="user",
            content=f"""Please review the following {language} code.
Focus on: {areas}

```{language}
{code}

Provide specific suggestions for improvement.""" ) ]


---

## 4. 依賴注入機制

### 4.1 Context 注入

FastMCP 提供 `Context` 物件，讓工具函數可以存取 MCP 功能：

```python
from fastmcp import FastMCP, Context

mcp = FastMCP("Demo")

@mcp.tool
async def process_data(data: str, ctx: Context) -> str:
    """Process data with progress reporting"""

    # 報告進度
    await ctx.report_progress(0.0, 1.0)

    # 發送日誌
    await ctx.log("Starting processing...", level="info")

    # 讀取其他資源
    config = await ctx.read_resource("config://settings")

    # 進行處理...
    result = data.upper()

    await ctx.report_progress(1.0, 1.0)
    return result

4.2 依賴注入實現原理

# dependencies.py

def _find_kwarg_by_type(fn: Callable, kwarg_type: type) -> str | None:
    """查找函數參數中類型匹配的參數名"""
    type_hints = get_type_hints(fn, include_extras=True)
    sig = inspect.signature(fn)

    for name, param in sig.parameters.items():
        annotation = type_hints.get(name, param.annotation)
        if is_class_member_of_type(annotation, kwarg_type):
            return name
    return None

@asynccontextmanager
async def resolve_dependencies(
    fn: Callable[..., Any],
    arguments: dict[str, Any]
) -> AsyncGenerator[dict[str, Any], None]:
    """解析依賴並注入 Context"""

    # 1. 過濾掉依賴參數 (安全措施)
    dependency_params = get_dependency_parameters(fn)
    user_args = {k: v for k, v in arguments.items()
                 if k not in dependency_params}

    # 2. 解析 Docket 依賴
    async with _resolve_fastmcp_dependencies(fn, user_args) as resolved:
        # 3. 注入 Context
        context_kwarg = _find_kwarg_by_type(fn, kwarg_type=Context)
        if context_kwarg and context_kwarg not in resolved:
            resolved[context_kwarg] = get_context()

        yield resolved

4.3 Docket 依賴

FastMCP 整合 Docket 進行背景任務處理：

from fastmcp.dependencies import CurrentDocket, CurrentWorker, Progress

@mcp.tool(task=True)  # 啟用背景任務
async def long_running_task(
    duration: int,
    progress: Progress = Progress(),  # 注入進度追蹤器
) -> str:
    """Execute a long-running task"""
    await progress.set_total(duration)

    for i in range(duration):
        await asyncio.sleep(1)
        await progress.increment()
        await progress.set_message(f"Step {i+1}/{duration}")

    return f"Completed in {duration} seconds"

5. 中間件系統

5.1 中間件架構

flowchart TD
    subgraph "Middleware Chain"
        A[Incoming Request] --> B[Middleware 1]
        B --> C[Middleware 2]
        C --> D[Middleware N]
        D --> E[Core Handler]
        E --> F[Response]
        F --> G[Middleware N]
        G --> H[Middleware 2]
        H --> I[Middleware 1]
        I --> J[Final Response]
    end

5.2 中間件基類

from fastmcp.server.middleware import Middleware, MiddlewareContext, CallNext

class Middleware:
    """中間件基類，使用分派處理器"""

    async def __call__(
        self,
        context: MiddlewareContext[T],
        call_next: CallNext[T, Any],
    ) -> Any:
        handler_chain = await self._dispatch_handler(context, call_next)
        return await handler_chain(context)

    async def _dispatch_handler(self, context, call_next) -> CallNext:
        """根據 method 分派到對應處理器"""
        handler = call_next

        match context.method:
            case "tools/call":
                handler = partial(self.on_call_tool, call_next=handler)
            case "resources/read":
                handler = partial(self.on_read_resource, call_next=handler)
            case "prompts/get":
                handler = partial(self.on_get_prompt, call_next=handler)
            # ... 更多方法

        return handler

    # 可覆寫的鉤子方法
    async def on_call_tool(self, context, call_next) -> ToolResult:
        return await call_next(context)

    async def on_read_resource(self, context, call_next):
        return await call_next(context)

5.3 自定義中間件範例

import time
import logging
from fastmcp.server.middleware import Middleware, MiddlewareContext

logger = logging.getLogger(__name__)

class LoggingMiddleware(Middleware):
    """記錄所有請求的中間件"""

    async def on_call_tool(self, context: MiddlewareContext, call_next):
        tool_name = context.message.name
        start = time.perf_counter()

        logger.info(f"Calling tool: {tool_name}")

        try:
            result = await call_next(context)
            elapsed = (time.perf_counter() - start) * 1000
            logger.info(f"Tool {tool_name} completed in {elapsed:.2f}ms")
            return result
        except Exception as e:
            logger.error(f"Tool {tool_name} failed: {e}")
            raise

class RateLimitMiddleware(Middleware):
    """速率限制中間件"""

    def __init__(self, max_requests: int = 100, window_seconds: int = 60):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.requests: dict[str, list[float]] = {}

    async def on_request(self, context, call_next):
        client_id = self._get_client_id(context)
        now = time.time()

        # 清理過期請求
        self.requests[client_id] = [
            t for t in self.requests.get(client_id, [])
            if now - t < self.window_seconds
        ]

        if len(self.requests[client_id]) >= self.max_requests:
            raise Exception("Rate limit exceeded")

        self.requests[client_id].append(now)
        return await call_next(context)

# 使用中間件
mcp = FastMCP(
    "My Server",
    middleware=[
        LoggingMiddleware(),
        RateLimitMiddleware(max_requests=50),
    ]
)

6. Provider 模式與伺服器組合

6.1 Provider 抽象

Provider 允許動態提供工具、資源和提示詞：

from fastmcp.server.providers import Provider
from fastmcp.tools.tool import Tool

class DatabaseToolProvider(Provider):
    """從資料庫動態載入工具"""

    def __init__(self, db_connection: str):
        self.db = Database(db_connection)

    async def list_tools(self) -> list[Tool]:
        rows = await self.db.fetch("SELECT * FROM tools WHERE enabled = 1")
        return [self._make_tool(row) for row in rows]

    async def get_tool(self, name: str) -> Tool | None:
        row = await self.db.fetchone(
            "SELECT * FROM tools WHERE name = ?", name
        )
        return self._make_tool(row) if row else None

    def _make_tool(self, row) -> Tool:
        return ConfigurableTool(
            name=row["name"],
            description=row["description"],
            parameters=json.loads(row["schema"]),
            operation=row["operation"],
        )

# 註冊 Provider
mcp = FastMCP("Dynamic Server")
mcp.add_provider(DatabaseToolProvider("sqlite:///tools.db"))

6.2 伺服器組合 (Mount)

FastMCP 支援將多個伺服器組合成一個：

from fastmcp import FastMCP

# 子應用：天氣服務
weather_app = FastMCP("Weather App")

@weather_app.tool
def get_weather(city: str) -> dict:
    """Get weather for a city"""
    return {"city": city, "temp": 25, "condition": "sunny"}

@weather_app.resource("weather://current")
def current_weather() -> dict:
    return {"temperature": 25, "humidity": 60}

# 子應用：新聞服務
news_app = FastMCP("News App")

@news_app.tool
def get_headlines() -> list[str]:
    """Get latest headlines"""
    return ["Breaking news 1", "Breaking news 2"]

# 主應用
main_app = FastMCP("Main Hub")

@main_app.tool
def system_status() -> dict:
    return {"status": "running", "uptime": "24h"}

# 掛載子應用（自動添加命名空間前綴）
main_app.mount(server=weather_app, prefix="weather")
main_app.mount(server=news_app, prefix="news")

# 結果：
# - weather_get_weather (tool)
# - news_get_headlines (tool)
# - system_status (tool)

6.3 ProxyProvider (遠程代理)

from fastmcp import FastMCP
from fastmcp.client import Client

# 代理遠程 MCP 伺服器
main_app = FastMCP("Proxy Hub")

remote_client = Client("http://remote-server:8000/mcp")
main_app.mount(server=remote_client, prefix="remote")

# 或使用 as_proxy 快速建立代理
proxy = FastMCP.as_proxy(
    Client("http://remote-server:8000/mcp"),
    name="Remote Proxy"
)

7. 傳輸層實現

7.1 支援的傳輸協議

graph TB
    subgraph "Client Transports"
        C1[StreamableHttpTransport]
        C2[SSETransport]
        C3[StdioTransport]
        C4[FastMCPTransport]
    end

    subgraph "Server Transports"
        S1[HTTP/Streamable HTTP]
        S2[SSE]
        S3[Stdio]
    end

    C1 <--> S1
    C2 <--> S2
    C3 <--> S3
    C4 <--> |In-Memory| S1

7.2 傳輸層抽象

# transports.py

class ClientTransport(abc.ABC):
    """客戶端傳輸抽象基類"""

    @abc.abstractmethod
    @contextlib.asynccontextmanager
    async def connect_session(
        self, **session_kwargs: Unpack[SessionKwargs]
    ) -> AsyncIterator[ClientSession]:
        """建立連接並產生 ClientSession"""
        raise NotImplementedError

class StreamableHttpTransport(ClientTransport):
    """Streamable HTTP 傳輸（推薦）"""

    def __init__(self, url: str, headers=None, auth=None):
        self.url = url
        self.headers = headers or {}
        self.auth = auth

    @contextlib.asynccontextmanager
    async def connect_session(self, **session_kwargs):
        async with streamable_http_client(self.url, ...) as transport:
            read_stream, write_stream, _ = transport
            async with ClientSession(
                read_stream, write_stream, **session_kwargs
            ) as session:
                yield session

class FastMCPTransport(ClientTransport):
    """進程內傳輸（用於測試）"""

    def __init__(self, fastmcp: FastMCP):
        self.fastmcp = fastmcp

    @contextlib.asynccontextmanager
    async def connect_session(self, **session_kwargs):
        # 使用內存流直接連接
        server_streams, client_streams = create_client_server_memory_streams()

        async with anyio.create_task_group() as tg:
            tg.start_soon(self._run_server, server_streams)
            async with ClientSession(*client_streams, **session_kwargs) as session:
                yield session

7.3 啟動伺服器

from fastmcp import FastMCP

mcp = FastMCP("My Server")

# Stdio 模式（用於 Claude Desktop）
mcp.run(transport="stdio")

# HTTP 模式（Streamable HTTP）
mcp.run(transport="http", host="0.0.0.0", port=8000)

# SSE 模式（舊版）
mcp.run(transport="sse", host="0.0.0.0", port=8000)

# 非同步版本
await mcp.run_async(transport="http", port=8000)

8. OAuth 認證系統

8.1 認證架構

sequenceDiagram
    participant Client as MCP Client
    participant Server as FastMCP Server
    participant OAuth as OAuth Provider

    Client->>Server: GET /.well-known/oauth-authorization-server
    Server-->>Client: OAuth Metadata

    Client->>Server: POST /authorize
    Server->>OAuth: Redirect to OAuth Provider
    OAuth-->>Client: Authorization Code

    Client->>Server: POST /token (code)
    Server->>OAuth: Exchange Code for Token
    OAuth-->>Server: Access Token
    Server-->>Client: Access Token

    Client->>Server: MCP Request + Bearer Token
    Server->>Server: Verify Token
    Server-->>Client: MCP Response

8.2 AuthProvider 層次

# auth.py

class AuthProvider(TokenVerifierProtocol):
    """認證提供者基類"""

    def __init__(self, base_url=None, required_scopes=None):
        self.base_url = base_url
        self.required_scopes = required_scopes or []

    async def verify_token(self, token: str) -> AccessToken | None:
        """驗證令牌 - 子類必須實現"""
        raise NotImplementedError

    def get_routes(self, mcp_path=None) -> list[Route]:
        """獲取認證路由"""
        return []

    def get_middleware(self) -> list:
        """獲取 HTTP 中間件"""
        return [
            Middleware(AuthenticationMiddleware, backend=BearerAuthBackend(self)),
            Middleware(AuthContextMiddleware),
        ]

class OAuthProvider(AuthProvider, OAuthAuthorizationServerProvider):
    """完整 OAuth 授權伺服器"""

    async def verify_token(self, token: str) -> AccessToken | None:
        return await self.load_access_token(token)

    def get_routes(self, mcp_path=None) -> list[Route]:
        # 創建 OAuth 路由 (/authorize, /token, /.well-known/...)
        return create_auth_routes(provider=self, ...)

8.3 內建認證提供者

提供者	Import Path
GitHub	`fastmcp.server.auth.providers.github.GitHubProvider`
Google	`fastmcp.server.auth.providers.google.GoogleProvider`
Azure	`fastmcp.server.auth.providers.azure.AzureProvider`
AWS Cognito	`fastmcp.server.auth.providers.aws.AWSCognitoProvider`
Discord	`fastmcp.server.auth.providers.discord.DiscordProvider`
WorkOS	`fastmcp.server.auth.providers.workos.WorkOSProvider`
Auth0	`fastmcp.server.auth.providers.auth0.Auth0Provider`

8.4 認證範例

伺服器端：

import os
from fastmcp import FastMCP
from fastmcp.server.auth.providers.github import GitHubProvider
from fastmcp.server.dependencies import get_access_token

auth = GitHubProvider(
    client_id=os.getenv("GITHUB_CLIENT_ID"),
    client_secret=os.getenv("GITHUB_CLIENT_SECRET"),
    base_url="http://localhost:8000",
)

mcp = FastMCP("Protected Server", auth=auth)

@mcp.tool
async def whoami() -> dict:
    """Get current user info"""
    token = get_access_token()
    return {
        "client_id": token.client_id,
        "scopes": token.scopes,
        "claims": token.claims,
    }

if __name__ == "__main__":
    mcp.run(transport="http", port=8000)

客戶端：

import asyncio
from fastmcp.client import Client

async def main():
    async with Client("http://localhost:8000/mcp", auth="oauth") as client:
        # Client 會自動處理 OAuth 流程
        result = await client.call_tool("whoami", {})
        print(result.data)

asyncio.run(main())

9. Context 與 Sampling

9.1 Context 功能總覽

@dataclass
class Context:
    """提供 MCP 功能訪問的上下文對象"""

    # === 基本功能 ===
    async def report_progress(self, progress: float, total: float | None = None):
        """報告進度"""

    async def log(self, message: str, level: LoggingLevel = "info"):
        """發送日誌到客戶端"""

    async def read_resource(self, uri: str) -> list[ResourceContent]:
        """讀取資源"""

    # === 狀態管理 ===
    def set_state(self, key: str, value: Any) -> None:
        """設置狀態"""

    def get_state(self, key: str) -> Any:
        """獲取狀態"""

    # === Sampling（LLM 調用）===
    async def sample(
        self,
        messages: str | Sequence[str | SamplingMessage],
        *,
        tools: Sequence[SamplingTool | Callable] | None = None,
        result_type: type[ResultT] | None = None,
        system_prompt: str | None = None,
        max_tokens: int = 1000,
    ) -> SamplingResult[ResultT]:
        """發送採樣請求並自動執行工具循環"""

9.2 Sampling 使用範例

基本文字生成：

from fastmcp import FastMCP, Context

mcp = FastMCP("AI Assistant")

@mcp.tool
async def write_haiku(topic: str, ctx: Context) -> str:
    """Write a haiku about any topic"""
    result = await ctx.sample(
        messages=f"Write a haiku about: {topic}",
        system_prompt="You are a poet. Write only the haiku, nothing else.",
        max_tokens=100,
    )
    return result.text or ""

結構化輸出：

from pydantic import BaseModel, Field

class SentimentAnalysis(BaseModel):
    sentiment: str  # "positive", "negative", "neutral"
    confidence: float
    keywords: list[str]
    explanation: str

@mcp.tool
async def analyze_sentiment(text: str, ctx: Context) -> dict:
    """Analyze text sentiment"""
    result = await ctx.sample(
        messages=f"Analyze the sentiment of: {text}",
        system_prompt="You are a sentiment analysis expert.",
        result_type=SentimentAnalysis,  # 強制結構化輸出
    )
    return result.result.model_dump()

帶工具的 Sampling：

def add(a: float, b: float) -> str:
    """Add two numbers"""
    return str(a + b)

def multiply(a: float, b: float) -> str:
    """Multiply two numbers"""
    return str(a * b)

@mcp.tool
async def smart_calculator(question: str, ctx: Context) -> str:
    """Answer math questions using tools"""
    result = await ctx.sample(
        messages=question,
        system_prompt="You are a helpful math assistant.",
        tools=[add, multiply],  # LLM 可以使用這些工具
    )
    return result.text or ""

10. 背景任務 (SEP-1686)

10.1 任務執行流程

sequenceDiagram
    participant Client as MCP Client
    participant Server as FastMCP Server
    participant Docket as Docket (Redis)
    participant Worker as Background Worker

    Client->>Server: call_tool (with _meta.task)
    Server->>Docket: Submit Task
    Docket-->>Server: Task ID
    Server-->>Client: CreateTaskResult (taskId, status: "running")

    Worker->>Docket: Poll for Tasks
    Docket-->>Worker: Task Data
    Worker->>Worker: Execute Tool Function
    Worker->>Docket: Store Result

    Client->>Server: tasks/get (taskId)
    Server->>Docket: Get Task Status
    Docket-->>Server: Task Result
    Server-->>Client: GetTaskResult (completed, result)

10.2 任務配置

from fastmcp import FastMCP
from fastmcp.dependencies import Progress
from fastmcp.server.tasks.config import TaskConfig
from typing import Annotated
from docket import Logged

mcp = FastMCP("Task Server")

# 方法 1：簡單啟用
@mcp.tool(task=True)
async def simple_task(data: str) -> str:
    await asyncio.sleep(5)
    return f"Processed: {data}"

# 方法 2：詳細配置
@mcp.tool(task=TaskConfig(mode="optional", poll_interval=timedelta(seconds=1)))
async def configurable_task(
    duration: Annotated[int, Logged],  # 參數會被記錄
    progress: Progress = Progress(),   # 注入進度追蹤
) -> str:
    await progress.set_total(duration)

    for i in range(duration):
        await asyncio.sleep(1)
        await progress.increment()
        await progress.set_message(f"Step {i+1}/{duration}")

    return f"Completed in {duration} seconds"

10.3 任務模式

模式	說明
`forbidden`	不支援背景任務（預設）
`optional`	支援但不強制，客戶端可選擇
`required`	強制背景執行

11. 錯誤處理機制

11.1 異常層次

# exceptions.py

class FastMCPError(Exception):
    """FastMCP 基礎異常"""

class ValidationError(FastMCPError):
    """參數或返回值驗證錯誤"""

class ResourceError(FastMCPError):
    """資源操作錯誤"""

class ToolError(FastMCPError):
    """工具操作錯誤"""

class PromptError(FastMCPError):
    """提示操作錯誤"""

class NotFoundError(Exception):
    """對象未找到"""

class DisabledError(Exception):
    """對象已禁用"""

11.2 錯誤遮罩

# 預設行為：遮罩內部錯誤細節
mcp = FastMCP("Server", mask_error_details=True)

# 內部實現：
async def _execute_tool(self, tool, tool_name, arguments) -> ToolResult:
    try:
        return await tool._run(arguments)
    except (ValidationError, PydanticValidationError):
        # 驗證錯誤永不遮罩
        raise
    except ToolError:
        raise
    except Exception as e:
        if self._mask_error_details:
            # 遮罩內部細節
            raise ToolError(f"Error calling tool {tool_name!r}") from e
        # 暴露完整錯誤
        raise ToolError(f"Error calling tool {tool_name!r}: {e}") from e

12. 測試指南

12.1 In-Memory 測試

import pytest
from fastmcp import FastMCP
from fastmcp.client import Client

# 建立伺服器
mcp = FastMCP("Test Server")

@mcp.tool
def add(a: int, b: int) -> int:
    return a + b

@mcp.resource("test://info")
def test_info() -> str:
    return "Test server info"

# 測試
@pytest.fixture
async def client():
    async with Client(mcp) as client:
        yield client

async def test_add_tool(client: Client):
    result = await client.call_tool("add", {"a": 2, "b": 3})
    assert result.data == 5

async def test_resource(client: Client):
    result = await client.read_resource("test://info")
    assert result[0].text == "Test server info"

@pytest.mark.parametrize("a,b,expected", [
    (0, 0, 0),
    (1, 1, 2),
    (-1, 1, 0),
])
async def test_add_parametrized(client: Client, a, b, expected):
    result = await client.call_tool("add", {"a": a, "b": b})
    assert result.data == expected

12.2 HTTP 測試

import httpx
import pytest

@pytest.fixture
async def server():
    mcp = FastMCP("Test")

    @mcp.tool
    def echo(msg: str) -> str:
        return msg

    # 啟動伺服器
    import asyncio
    task = asyncio.create_task(
        mcp.run_http_async(host="127.0.0.1", port=8765)
    )
    await asyncio.sleep(0.5)  # 等待啟動
    yield
    task.cancel()

async def test_http_endpoint(server):
    async with httpx.AsyncClient() as client:
        response = await client.get("http://127.0.0.1:8765/health")
        assert response.status_code == 200

13. FastAPI 整合

13.1 從 FastAPI 建立 MCP

from fastapi import FastAPI
from fastmcp import FastMCP
from fastmcp.server.openapi import MCPType, RouteMap

# 現有的 FastAPI 應用
app = FastAPI()

@app.get("/users", tags=["users"])
async def get_users():
    return [{"id": 1, "name": "Alice"}]

@app.post("/users", tags=["users", "admin"])
async def create_user(name: str):
    return {"id": 2, "name": name}

@app.get("/health", tags=["public"])
async def health():
    return {"status": "ok"}

# 轉換為 MCP 伺服器
mcp = FastMCP.from_fastapi(
    app=app,
    route_maps=[
        # admin 標籤的路由轉為 Tool
        RouteMap(methods="*", pattern=r".*", mcp_type=MCPType.TOOL, tags={"admin"}),
        # GET 請求轉為 Resource
        RouteMap(methods=["GET"], pattern=r".*", mcp_type=MCPType.RESOURCE),
        # POST 請求轉為 Tool
        RouteMap(methods=["POST"], pattern=r".*", mcp_type=MCPType.TOOL),
    ],
)

13.2 ASGI 整合

from starlette.applications import Starlette
from starlette.routing import Mount

# FastMCP 應用
mcp = FastMCP("My Server")
mcp_app = mcp.http_app(path="/mcp")

# 整合到現有 ASGI 應用
app = Starlette(
    routes=[
        Mount("/api/mcp", app=mcp_app),
        # ... 其他路由
    ],
    lifespan=mcp_app.lifespan,  # 重要：傳遞 lifespan
)

14. 配置檔案

14.1 fastmcp.json 格式

{
  "$schema": "https://gofastmcp.com/public/schemas/fastmcp.json/v1.json",
  "source": {
    "path": "server.py",
    "entrypoint": "mcp"
  },
  "environment": {
    "python": "3.12",
    "dependencies": ["requests>=2.31.0", "httpx"],
    "requirements": null,
    "project": null
  },
  "deployment": {
    "transport": "http",
    "host": "127.0.0.1",
    "port": 8000,
    "path": "/mcp/",
    "log_level": "INFO",
    "env": {
      "DEBUG": "false",
      "API_TIMEOUT": "30"
    }
  }
}

14.2 簡化版本

{
  "$schema": "https://gofastmcp.com/public/schemas/fastmcp.json/v1.json",
  "entrypoint": "server.py",
  "environment": {
    "dependencies": ["pydantic", "httpx"]
  }
}

15. 常見 Prompt 範例

15.1 程式碼審查 Prompt

@mcp.prompt("code_review")
def code_review(
    language: str,
    code: str,
    focus: str = "general quality"
) -> str:
    return f"""Please review the following {language} code.

Focus areas: {focus}

```{language}
{code}

Provide:

Summary of what the code does
Potential issues or bugs
Suggestions for improvement
Security considerations (if applicable)
Performance optimization opportunities"""


### 15.2 資料分析 Prompt

```python
from fastmcp.prompts.prompt import Message

@mcp.prompt("analyze_data")
def analyze_data(
    data_description: str,
    analysis_goals: list[str],
    output_format: str = "markdown"
) -> list[Message]:
    goals_text = "\n".join(f"- {goal}" for goal in analysis_goals)

    return [
        Message(
            role="user",
            content=f"""You are a data analyst. Please analyze the following data.

## Data Description
{data_description}

## Analysis Goals
{goals_text}

## Output Format
Please provide your analysis in {output_format} format.

Include:
1. Executive Summary
2. Key Findings
3. Statistical Analysis (if applicable)
4. Visualizations (describe what charts would be helpful)
5. Recommendations
6. Limitations and Caveats"""
        )
    ]

15.3 多語言翻譯 Prompt

@mcp.prompt("translate")
def translate(
    text: str,
    source_lang: str = "auto",
    target_lang: str = "en",
    style: str = "formal"
) -> str:
    return f"""Translate the following text.

Source language: {source_lang}
Target language: {target_lang}
Style: {style}

Text to translate:
---
{text}
---

Provide:
1. The translation
2. Alternative translations for key phrases (if applicable)
3. Cultural notes (if relevant)"""

16. 優勢與特點

16.1 技術優勢

特點	說明
型別安全	基於 Pydantic 的完整型別驗證
自動文檔	從 docstring 和型別註解自動生成 JSON Schema
非同步優先	基於 asyncio 和 anyio 的完整非同步支援
可擴展性	Provider 和 Middleware 模式支援無限擴展
測試友好	In-Memory Transport 支援無網路測試
生產就緒	內建認證、速率限制、錯誤處理

16.2 與其他框架比較

FastMCP vs 官方 MCP SDK：
├── FastMCP: 高階抽象，裝飾器驅動，適合快速開發
└── MCP SDK: 底層 API，完全控制，適合特殊需求

FastMCP vs LangChain/LangGraph：
├── FastMCP: 專注於 MCP 協議，工具/資源/提示詞
├── LangChain: 通用 LLM 應用框架，Chain/Agent
└── LangGraph: 狀態機和工作流程

17. 潛在問題與注意事項

17.1 效能考量

Pydantic 序列化開銷：每次請求都會進行 JSON Schema 驗證
中間件鏈長度：過多中間件會增加延遲
背景任務 Redis 依賴：SEP-1686 任務需要 Redis

17.2 安全注意

錯誤遮罩：生產環境應啟用 mask_error_details=True
OAuth 配置：確保正確設置 redirect_uri 和 base_url
依賴注入安全：用戶無法覆寫注入的參數（框架層級保護）

17.3 相容性

Python 版本：需要 Python 3.10+
MCP SDK 版本：需與官方 SDK 版本匹配
Pydantic 版本：需要 Pydantic v2

17.4 常見陷阱

# 錯誤：同步函數用於背景任務
@mcp.tool(task=True)
def sync_task(data: str) -> str:  # 會報錯！
    return data

# 正確：必須使用 async
@mcp.tool(task=True)
async def async_task(data: str) -> str:
    return data

# 錯誤：忘記傳遞 lifespan（ASGI 整合）
app = Starlette(routes=[Mount("/mcp", mcp_app)])  # 會導致任務失敗

# 正確：傳遞 lifespan
app = Starlette(routes=[Mount("/mcp", mcp_app)], lifespan=mcp_app.lifespan)

18. 總結

FastMCP 是一個設計精良的 Python MCP 框架，其核心設計原則包括：

裝飾器驅動：簡潔的 @mcp.tool、@mcp.resource、@mcp.prompt API
型別優先：完整的 Pydantic 整合和自動 Schema 生成
可組合性：Provider、Middleware、Mount 支援模組化架構
生產就緒：內建認證、錯誤處理、背景任務
測試友好：In-Memory Transport 支援快速測試

對於需要建構 MCP 服務的 Python 開發者，FastMCP 提供了一個高效、類型安全、功能完整的解決方案。