你当前正在访问 Microsoft Azure Global Edition 技术文档网站。 如果需要访问由世纪互联运营的 Microsoft Azure 中国技术文档网站,请访问 https://docs.azure.cn

教程:将 Python MCP 服务器部署到 Azure 容器应用

在本教程中,你将生成一个模型上下文协议(MCP)服务器,该服务器使用 FastAPI 和 MCP Python SDK 公开任务管理工具。 将服务器部署到 Azure 容器应用,并从 VS Code 中的 GitHub Copilot 聊天连接到它。

在本教程中,你将了解:

  • 创建公开 MCP 工具的 FastAPI 应用
  • 使用 GitHub Copilot 在本地测试 MCP 服务器
  • 容器化应用并将其部署到 Azure 容器应用
  • 将 GitHub Copilot 连接到部署的 MCP 服务器

先决条件

创建应用基架

在本部分中,你将使用 FastAPI 和 MCP Python SDK 创建新的 Python 项目。

  1. 创建项目目录并设置虚拟环境:

    mkdir tasks-mcp-server && cd tasks-mcp-server
    python -m venv .venv
    source .venv/bin/activate
    
  2. 创建 requirements.txt

    fastapi>=0.115.0
    uvicorn>=0.30.0
    mcp[cli]>=1.2.0
    
  3. 安装依赖项:

    pip install -r requirements.txt
    
  4. 为内存中数据存储创建 task_store.py

    from dataclasses import dataclass, field
    from datetime import datetime, timezone
    
    
    @dataclass
    class TaskItem:
        id: int
        title: str
        description: str
        is_complete: bool = False
        created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    
        def to_dict(self) -> dict:
            return {
                "id": self.id,
                "title": self.title,
                "description": self.description,
                "is_complete": self.is_complete,
                "created_at": self.created_at.isoformat(),
            }
    
    
    class TaskStore:
        def __init__(self):
            self._tasks: list[TaskItem] = [
                TaskItem(1, "Buy groceries", "Milk, eggs, bread"),
                TaskItem(2, "Write docs", "Draft the MCP tutorial", True),
            ]
            self._next_id = 3
    
        def get_all(self) -> list[dict]:
            return [t.to_dict() for t in self._tasks]
    
        def get_by_id(self, task_id: int) -> dict | None:
            task = next((t for t in self._tasks if t.id == task_id), None)
            return task.to_dict() if task else None
    
        def create(self, title: str, description: str) -> dict:
            task = TaskItem(self._next_id, title, description)
            self._next_id += 1
            self._tasks.append(task)
            return task.to_dict()
    
        def toggle_complete(self, task_id: int) -> dict | None:
            task = next((t for t in self._tasks if t.id == task_id), None)
            if task is None:
                return None
            task.is_complete = not task.is_complete
            return task.to_dict()
    
        def delete(self, task_id: int) -> bool:
            task = next((t for t in self._tasks if t.id == task_id), None)
            if task is None:
                return False
            self._tasks.remove(task)
            return True
    
    
    # For demonstration only — not thread-safe.
    store = TaskStore()
    

    数据 TaskItem 类使用 to_dict() 序列化方法定义数据模型。 TaskStore 类管理一个在内存中预填充示例数据的列表,并提供 CRUD 方法。 为了简单起见,模块级 store 单一实例在应用程序中共享。

定义 MCP 工具

在本部分中,你将定义 AI 模型可以在 FastAPI 应用程序中调用和装载 MCP 服务器的 MCP 工具。

  1. 创建 mcp_server.py

    from mcp.server.fastmcp import FastMCP
    from task_store import store
    
    mcp = FastMCP("TasksMCP", stateless_http=True)
    
    
    @mcp.tool()
    async def list_tasks() -> list[dict]:
        """List all tasks with their ID, title, description, and completion status."""
        return store.get_all()
    
    
    @mcp.tool()
    async def get_task(task_id: int) -> dict | None:
        """Get a single task by its numeric ID.
    
        Args:
            task_id: The numeric ID of the task to retrieve.
        """
        return store.get_by_id(task_id)
    
    
    @mcp.tool()
    async def create_task(title: str, description: str) -> dict:
        """Create a new task with the given title and description. Returns the created task.
    
        Args:
            title: A short title for the task.
            description: A detailed description of what the task involves.
        """
        return store.create(title, description)
    
    
    @mcp.tool()
    async def toggle_task_complete(task_id: int) -> str:
        """Toggle a task's completion status between complete and incomplete.
    
        Args:
            task_id: The numeric ID of the task to toggle.
        """
        task = store.toggle_complete(task_id)
        if task:
            status = "complete" if task["is_complete"] else "incomplete"
            return f"Task {task['id']} is now {status}."
        return f"Task with ID {task_id} not found."
    
    
    @mcp.tool()
    async def delete_task(task_id: int) -> str:
        """Delete a task by its numeric ID.
    
        Args:
            task_id: The numeric ID of the task to delete.
        """
        if store.delete(task_id):
            return f"Task {task_id} deleted."
        return f"Task with ID {task_id} not found."
    

    要点:

    • FastMCP("TasksMCP", stateless_http=True) 使用 Python SDK 中的无状态 HTTP 模式创建 MCP 服务器。 可流式传输的 HTTP 终结点默认为 /mcp 子路径。
    • 每个 @mcp.tool() 函数将成为可调用的工具。 函数 docstring 和参数注释可帮助 AI 模型了解如何使用每个工具。
  2. 创建 app.py。 此文件定义装载 MCP 服务器的 FastAPI 应用程序:

    from contextlib import AsyncExitStack, asynccontextmanager
    
    from fastapi import FastAPI
    from fastapi.responses import JSONResponse
    
    from mcp_server import mcp
    
    
    @asynccontextmanager
    async def lifespan(app: FastAPI):
        async with AsyncExitStack() as stack:
            await stack.enter_async_context(mcp.session_manager.run())
            yield
    
    
    app = FastAPI(lifespan=lifespan)
    app.mount("/", mcp.streamable_http_app())
    
    
    @app.get("/health")
    async def health():
        return JSONResponse({"status": "healthy"})
    

    MCP 服务器应用装载在根位置(/)。 SDK 的可流式传输 HTTP 终结点默认为 /mcp,因此完整的终结点路径为 /mcp

    单独的 /health 终结点用于容器应用运行状况探测。 MCP 终结点需要 JSON-RPC POST 请求,不适合作为运行状况检查。

在本地测试 MCP 服务器

在部署到 Azure 之前,请验证 MCP 服务器是否正常工作,方法是在本地运行它并从 GitHub Copilot 进行连接。

  1. 启动应用程序:

    uvicorn app:app --reload --port 8080
    
  2. 打开 VS Code,然后打开 Copilot 对话助手 并选择 代理 模式。

  3. 选择 “工具 ”按钮,然后选择“ 添加更多工具...”>添加 MCP 服务器

  4. 选择“HTTP(HTTP 或 Server-Sent 事件)”。

  5. 输入服务器 URL: http://localhost:8080/mcp

  6. 输入服务器 ID: tasks-mcp

  7. 选择 工作区设置

  8. 在新的 Copilot 对话助手 提示符中,键入 :“显示我所有任务”

  9. 当 Copilot 提示您需要确认 MCP 工具时,选择继续

您应该能够看到从内存存储返回的任务列表。

小窍门

尝试其他提示,例如“创建任务以查看 PR”、“将任务 1 标记为已完成”或“删除任务 2”。

容器化应用程序

将应用程序打包为 Docker 容器,以便在部署到 Azure 之前在本地对其进行测试。

  1. 创建 Dockerfile

    FROM python:3.12-slim
    
    WORKDIR /app
    
    COPY requirements.txt .
    RUN pip install --no-cache-dir -r requirements.txt
    
    COPY . .
    
    EXPOSE 8080
    CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8080"]
    

    Dockerfile 使用 Python 3.12 精简基础映像,从 requirements.txt中安装依赖项,然后复制应用程序代码。 Uvicorn 在端口 8080 上为 FastAPI 应用提供服务。

  2. 验证容器生成并在本地运行:

    docker build -t tasks-mcp-server .
    docker run -p 8080:8080 tasks-mcp-server
    

    确认运行状况终结点有响应: curl http://localhost:8080/health

部署到 Azure 容器应用

容器化应用程序后,使用 Azure CLI 将其部署到 Azure 容器应用。 该 az containerapp up 命令在云中生成容器映像,因此,此步骤不需要计算机上的 Docker。

  1. 设置环境变量:

    RESOURCE_GROUP="mcp-tutorial-rg"
    LOCATION="eastus"
    ENVIRONMENT_NAME="mcp-env"
    APP_NAME="tasks-mcp-server-py"
    
  2. 创建资源组和容器应用环境:

    az group create --name $RESOURCE_GROUP --location $LOCATION
    
    az containerapp env create \
        --name $ENVIRONMENT_NAME \
        --resource-group $RESOURCE_GROUP \
        --location $LOCATION
    
  3. 部署容器应用:

    az containerapp up \
        --name $APP_NAME \
        --resource-group $RESOURCE_GROUP \
        --environment $ENVIRONMENT_NAME \
        --source . \
        --ingress external \
        --target-port 8080
    
  4. 配置 CORS:

    az containerapp ingress cors enable \
        --name $APP_NAME \
        --resource-group $RESOURCE_GROUP \
        --allowed-origins "*" \
        --allowed-methods "GET,POST,DELETE,OPTIONS" \
        --allowed-headers "*"
    

    注释

    在生产环境中,请将通配符源替换为具体的受信任源。 请参阅 容器应用中的安全 MCP 服务器

  5. 验证部署:

    APP_URL=$(az containerapp show \
        --name $APP_NAME \
        --resource-group $RESOURCE_GROUP \
        --query "properties.configuration.ingress.fqdn" -o tsv)
    
    curl https://$APP_URL/health
    

将 GitHub Copilot 连接到已部署的服务器

现在 MCP 服务器在 Azure 中运行,请将 VS Code 配置为将 GitHub Copilot 连接到已部署的终结点。

  1. 创建或更新 .vscode/mcp.json

    {
        "servers": {
            "tasks-mcp-server": {
                "type": "http",
                "url": "https://<your-app-fqdn>/mcp"
            }
        }
    }
    

    <your-app-fqdn> 替换为部署输出中的 FQDN。

  2. 在 VS Code 中,在代理模式下打开 Copilot 聊天。

  3. 验证 tasks-mcp-server 是否显示在“工具”列表中。 根据需要选择 “开始 ”。

  4. 使用“ 创建任务以部署过渡环境”等提示进行测试。

配置缩放以供交互式使用

默认情况下,Azure 容器应用可以扩展到零个副本。 对于为 Copilot 等交互式客户端提供服务的 MCP 服务器,冷启动会导致明显的延迟。 设置最小副本计数以保持至少一个实例运行:

az containerapp update \
    --name $APP_NAME \
    --resource-group $RESOURCE_GROUP \
    --min-replicas 1

安全注意事项

本教程为简单起见,使用未经身份验证的 MCP 服务器。 在生产环境中运行 MCP 服务器之前,请查看以下建议。 当由大型语言模型(LLM)提供支持的代理调用 MCP 服务器时,请注意 提示注入 攻击。

  • 身份验证和授权:使用 Microsoft Entra ID 保护 MCP 服务器。 请参阅 容器应用中的安全 MCP 服务器
  • 输入验证:始终验证工具参数。 使用 Pydantic 对工具输入强制实施数据验证。
  • HTTPS:Azure 容器应用默认使用自动 TLS 证书强制实施 HTTPS。
  • 最低特权:仅公开用例所需的工具。 避免使用在未经确认的情况下执行破坏性操作的工具。
  • CORS:将允许的源限制为生产中的受信任域。
  • 日志记录和监视:记录 MCP 工具调用进行审核。 使用 Azure Monitor 和 Log Analytics。

清理资源

如果不打算继续使用此应用程序,请删除资源组以删除在本教程中创建的所有资源:

az group delete --resource-group $RESOURCE_GROUP --yes --no-wait

后续步骤