掘金 人工智能 08月21日
MCP 实战——MCP 服务器的身份验证与部署
index_new5.html
../../../zaker_core/zaker_tpl_static/wap/tpl_guoji1.html

 

本章详细介绍了如何将MCP服务器从开发原型升级为可用于生产的环境。重点讲解了使用JWT Bearer Token进行身份验证,以保护服务器免受未授权访问。同时,通过访问令牌声明,实现了对客户端的细粒度授权。在部署方面,提供了两种主流方案:一是利用Docker将服务器容器化,并部署到Google Cloud Run等现代无服务器平台,以实现高可用性和可扩展性;二是部署到传统的虚拟机环境,并结合Nginx作为反向代理,以获得更大的控制权。通过这些步骤,用户可以构建、加固并成功部署健壮的、面向真实场景的MCP应用。

🛡️ **JWT Bearer Token身份验证**:文章详细介绍了如何通过JWT Bearer Token来保护MCP服务器,确保只有授权的客户端才能访问。这种方法通过数字签名的令牌来验证客户端身份,服务器仅需公钥即可验证令牌的有效性,从而实现了安全且易于扩展的访问控制。

🔑 **访问令牌声明与细粒度授权**:在服务器的受保护资源内部,可以通过`get_access_token()`函数访问已验证的令牌对象,获取客户端的身份信息(如`client_id`或`subject`)以及其被授予的权限(`scopes`)。这使得可以实现基于这些信息的细粒度授权逻辑,例如根据不同的用户或权限执行不同的操作。

🐳 **Docker化与无服务器部署**:文章指导用户如何使用Dockerfile将MCP服务器应用容器化,以实现跨平台的可移植性。随后,详细阐述了如何将该容器镜像部署到Google Cloud Run等无服务器平台,该平台能够自动处理服务器的启动、停止、扩展和负载均衡,用户只需为实际用量付费,大大简化了运维工作。

⚙️ **虚拟机与Nginx反向代理部署**:对于需要更高控制权的用户,文章提供了在传统虚拟机上部署MCP服务器的方案。通过使用`screen`等工具使Python服务器在后台常驻运行,并配置Nginx作为反向代理,监听标准端口并将流量转发给MCP服务器,从而实现对外提供服务,并能完全控制操作系统和网络环境。

在上一章中,你深入探索了 MCP 的核心,把服务器从简单的“命令—响应”系统升级为能与 LLM 协作的上下文感知伙伴。你学会了用 Resources 暴露只读数据、用 Prompts 引导 LLM,并借助 Context 对象实现日志、进度汇报与引导提问(elicitation)的交互。

可以说,你已经为 AI 打造了一间功能强大的“互动工坊”。但现在,这间工坊既没有上锁,又运行在临时车库(你的本机)里。这对开发很完美;可如果要构建真实世界的应用,你需要安全稳定的托管环境

本章的目标,是把你的 MCP 服务器从原型晋级为可用于生产的服务。我们将解决上线最关键的两个方面:安全(只允许授权客户端使用你的服务器)与部署(将其运行在稳健、可扩展、可 7×24 运行的环境中)。

在本章结束时,你将能够:

让我们先“给门上锁”,再把工坊搬到“黄金地段”。

身份验证:守住你的工具

到目前为止,任何找到你服务器 URL 的人都能调用其工具。对一个“字符计数”工具也许无伤大雅;但如果工具会访问私有数据库代用户发邮件、或调用付费 API 呢?不受限的访问将是安全与成本的双重灾难。

**身份验证(Authentication)**就是验证客户端“是谁”。对 MCP 服务器,我们采用一种现代、标准的方法:JWT Bearer Token 身份验证

基本思路如下:

这种方式既安全又易扩展。你的 MCP 服务器不处理密码或机密,只需验证签名令牌。下面分别用 fastmcpmcp 来实现。

保护 fastmcp 服务器

fastmcp 对 Bearer Token 的配置非常简洁:既提供验证令牌的辅助类,也提供(仅用于开发演示的)创建令牌的工具。

让我们构建一个带受保护资源的服务器。

图 54. walled_mcp_server.py(fastmcp)

from fastmcp import FastMCPfrom fastmcp.server.auth.providers.bearer import RSAKeyPairfrom fastmcp.server.auth import BearerAuthProviderfrom fastmcp.server.dependencies import get_access_token, AccessToken# 1. Generate a public/private key pair for signing/verifying tokens.# In production, the private key lives on an Authorization Server.key_pair = RSAKeyPair.generate()# 2. Configure the authentication provider.# The server only needs the public key to verify tokens.auth = BearerAuthProvider(    public_key=key_pair.public_key,    issuer="https://your-awesome-mcp-server.com",    audience="my-mcp-server")# 3. Attach the auth provider to the MCP server instance.mcp = FastMCP("My MCP Server", auth=auth)# 4. For demonstration, create a valid token signed with the private key.token = key_pair.create_token(    subject="mcp-user",    issuer="https://your-awesome-mcp-server.com",    audience="my-mcp-server",    scopes=["read", "write"])# 5. Save the token for our client to use.with open("token.txt", "w") as f:    f.write(token)# 6. Define a protected resource.@mcp.resource("data://database")def get_data() -> str:    # 7. Access the validated token's claims inside the function.    access_token: AccessToken = get_access_token()    print(f"Access granted! Scopes: {access_token.scopes}, Subject: {access_token.client_id}")    return "Secret data from the database"if __name__ == "__main__":    mcp.run(transport="http", port=9000)

逐点说明:

接着是客户端:读取并携带令牌访问服务器。

图 55. walled_mcp_client.py(fastmcp)

import asynciofrom fastmcp import Clientfrom fastmcp.client.transports import (    StreamableHttpTransport,)from pathlib import Path# 1. Read the token from the file.token = Path("token.txt").read_text().strip()# 2. Pass the token to the Client constructor.client = Client(transport=StreamableHttpTransport("http://localhost:9000/mcp"),                auth=token)async def main():    async with client:        data = await client.read_resource("data://database")        print(data)asyncio.run(main())

客户端逻辑非常直接:

试运行:
图 56. 终端

> python walled_mcp_server.py

再在新终端运行客户端:
图 57. 终端

> python walled_mcp_client.py[TextResourceContents(uri=AnyUrl('data://database'), mimeType='text/plain', meta=None, text='Secret data from the database')]

成功!服务器验证了令牌,并在 get_data 中可读取到其 claims。若去掉客户端的 auth=token,服务器将返回 401 Unauthorized

保护 mcp 库服务器

如你所料,用更底层的 mcp 库实现会更显式,但原理完全相同:你需要手工搭建 fastmcp 已帮你封装好的组件。

图 58. walled_mcp_server.py(mcp)

import timefrom typing import Anyfrom jose import jwtfrom cryptography.hazmat.primitives import serializationfrom cryptography.hazmat.primitives.asymmetric import rsafrom mcp.server.fastmcp.server import FastMCPfrom mcp.server.auth.provider import TokenVerifier, AccessTokenfrom mcp.server.auth.middleware.auth_context import get_access_tokenfrom mcp.server.auth.settings import AuthSettings# 1. Create a custom token verifier class.class SimpleJWTVerifier(TokenVerifier):    def __init__(self, public_key: str, audience: str, issuer: str):        self.public_key = public_key        self.audience = audience        self.issuer = issuer    async def verify_token(self, token: str) -> AccessToken | None:        try:            payload = jwt.decode(                token,                self.public_key,                algorithms=["RS256"],                audience=self.audience,                issuer=self.issuer,            )            return AccessToken(                token=token,                client_id=payload.get("sub"),                scopes=payload.get("scopes", []),                expires_at=payload.get("exp"),            )        except jwt.JWTError:            return None# 2. Manually generate RSA keys using the cryptography library.private_key_obj = rsa.generate_private_key(public_exponent=65537, key_size=2048)private_key_pem = private_key_obj.private_bytes(    encoding=serialization.Encoding.PEM,    format=serialization.PrivateFormat.PKCS8,    encryption_algorithm=serialization.NoEncryption(),).decode("utf-8")public_key_pem = private_key_obj.public_key().public_bytes(    encoding=serialization.Encoding.PEM,    format=serialization.PublicFormat.SubjectPublicKeyInfo,).decode("utf-8")# 3. Instantiate the verifier and configure auth settings.token_verifier = SimpleJWTVerifier(    public_key=public_key_pem,    audience="my-mcp-server",    issuer="https://your-awesome-mcp-server.com",)auth_settings = AuthSettings(    issuer_url="https://auth.my-mcp-server.com",    resource_server_url="http://localhost:9000",    required_scopes=["data:read"])# 4. Attach the verifier and settings to the server.mcp = FastMCP(    name="My Simple SDK Server",    token_verifier=token_verifier,    auth=auth_settings,    host="127.0.0.1",    port=9000)# 5. Manually create the token claims and encode it using jose.claims: dict[str, Any] = {    "iss": "https://your-awesome-mcp-server.com",    "aud": "my-mcp-server",    "sub": "mcp-user",    "exp": int(time.time()) + 3600,    "iat": int(time.time()),    "scopes": ["data:read", "data:write"],}token = jwt.encode(claims, private_key_pem, algorithm="RS256")with open("token.txt", "w") as f:    f.write(token)@mcp.resource("data://database")def get_data() -> str:    access_token: AccessToken = get_access_token()    print(f"Scopes of token: {access_token.scopes}")    print(f"Client id or subject: {access_token.client_id}")    return "Secret data from the database"if __name__ == "__main__":    mcp.run(transport="streamable-http")

差异要点:

在此之前,先安装依赖库:
图 59. 安装依赖

> uv add python-jose[cryptography]

客户端也更“手工化”,需要自行构造 Authorization 请求头。

图 60. walled_mcp_client.py(mcp)

import asynciofrom pathlib import Pathfrom datetime import timedeltafrom mcp.client.streamable_http import streamablehttp_clientfrom mcp.client.session import ClientSessionSERVER_URL = "http://localhost:9000/mcp"async def main():    token = Path("token.txt").read_text().strip()    # Manually create the headers dictionary.    auth_headers = {"Authorization": f"Bearer {token}"}    # Pass the headers to the client transport.    async with streamablehttp_client(        url=SERVER_URL,        headers=auth_headers,        timeout=timedelta(seconds=30)    ) as (read_stream, write_stream, get_session_id):        async with ClientSession(read_stream, write_stream) as session:            await session.initialize()            data = await session.read_resource("data://database")            print(data)if __name__ == "__main__":    asyncio.run(main())

关闭先前的 fastmcp 服务器,运行新的 mcp 版本服务器;再运行该客户端。结果与 fastmcp 相同,说明两种抽象层级都能达到同样安全的效果。

图 61. 终端

> python walled_mcp_client.pymeta=None contents=[TextResourceContents(uri=AnyUrl('data://database'), mimeType='text/plain', meta=None, text='Secret data from the database')]

至此,你已经掌握了保护 MCP 服务器的基础。接下来,是时候把它们从本地“搬出去”,走向真实的线上环境了。

部署:正式上线(Going Live)

在本机终端里跑一个服务器适合开发阶段,但并不是现实世界的解决方案。生产部署意味着让你的代码运行在可靠、可扩展、始终可用的服务器上。我们将探讨两种常见策略:把应用部署到现代无服务器平台(Google Cloud Run),以及部署到传统虚拟机(VM)

使用 Docker 与 Cloud Run 的无服务器部署

Google Cloud Run、AWS Lambda、Azure Functions 这样的无服务器平台是部署应用的绝佳方式。你提供容器中的代码,平台负责剩下的一切:启动服务器、在无流量时自动停止、在高并发时自动扩容。你只为实际用量付费。

我们将把一个简单的 fastmcp 服务器部署到 Google Cloud Run。流程分四步:创建项目文件构建 Docker 镜像部署到 Cloud Run测试

第 1 步:创建项目文件

先为部署项目创建一个新目录。在目录内创建以下四个文件。

应用(server.py)

这是一个最简的 MCP 服务器。注意 host 与 port 的设置。

图 62. server.py

from fastmcp import FastMCP mcp = FastMCP("MCP Server on Cloud Run")@mcp.tool()def count_characters(string: str) -> int:    return len(string)if __name__ == "__main__":    # Host '0.0.0.0' listens on all network interfaces, which is required for containers.    # Cloud Run provides the port via the $PORT environment variable, which Uvicorn uses automatically.    # 8080 is a common default.    mcp.run(transport="http", host="0.0.0.0", port=8080)

依赖(pyproject.toml)

告诉 uv 需要安装哪些库。

图 63. pyproject.toml

[project]name = "mcp-on-cloudrun"version = "0.1.0"description = "MCP on Cloud Run"requires-python = ">=3.10"dependencies = [    "fastmcp==2.10.5",]

容器构建脚本(Dockerfile)

Dockerfile 是构建容器镜像的指令集。下面这个示例使用现代的多阶段构建与 uv,兼顾速度与体积

图 64. Dockerfile

# Start with a small, official Python image.FROM python:3.12-slim# Use a multi-stage build to copy the `uv` binary without its build environment.COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/# Copy our application code into the image.COPY . /appWORKDIR /app# Install dependencies using the copied `uv` binary.RUN uv sync# Tell Docker which port the container will listen on.EXPOSE $PORT# The command to run when the container starts.CMD ["uv", "run", "server.py"]

测试客户端(client.py)

稍后用于测试部署;它会连接到本地代理。

图 65. client.py

import asynciofrom fastmcp import Clientfrom fastmcp.client.transports import (    StreamableHttpTransport,)client = Client(transport=StreamableHttpTransport("http://localhost:8080/mcp"))async def main():    async with client:        tools = await client.list_tools()        print(tools)                output = await client.call_tool("count_characters", {"string": "Strawberry is delicious!"})        extracted_text = output.content[0].text        print(extracted_text)asyncio.run(main())

第 2 步:构建并部署到 Google Cloud

接下来使用 gcloud 命令行工具构建镜像并部署。确保你已安装并配置好 Google Cloud SDK

首先在 Artifact Registry 中创建一个仓库来存放 Docker 镜像。把 your-gcp-project 替换为你的实际 GCP 项目 ID。

图 66. 终端

> gcloud artifacts repositories create mcp-servers --repository-format=docker --location=asia-southeast1Created repository [mcp-servers].

使用 Cloud Build 按 Dockerfile 构建镜像并推送到刚创建的仓库。

图 67. 终端

> gcloud builds submit --region=asia-southeast1 --tag asia-southeast1-docker.pkg.dev/your-gcp-project/mcp-servers/mcp-server:latest...IMAGES: asia-southeast1-docker.pkg.dev/your-gcp-project/mcp-servers/mcp-server (+1 more)STATUS: SUCCESS

该命令会打包项目文件、发送到 Google Cloud Build,随后按 Dockerfile 的步骤构建镜像并保存。

权限提示:首次使用 Cloud Run 时,可能需要为你的账号授予管理部署及以服务账号身份运行的权限。通常每个项目只需执行一次。将 your-gcp-projectyouremail@googlecloud.com 替换为你的信息。

gcloud projects add-iam-policy-binding your-gcp-project --member="user:youremail@googlecloud.com" --role="roles/run.admin"# 获取项目编号gcloud projects describe your-gcp-project --format='value(projectNumber)'# 使用项目编号为默认计算服务账号授予 Service Account User 角色gcloud iam service-accounts add-iam-policy-binding "PROJECT_NUMBER-compute@developer.gserviceaccount.com" --member="user:youremail@googlecloud.com" --role="roles/iam.serviceAccountUser"

最后,部署镜像到 Cloud Run--no-allow-unauthenticated 使服务默认私有,这符合安全最佳实践。

图 68. 终端

> gcloud run deploy mcp-server --image asia-southeast1-docker.pkg.dev/your-gcp-project/mcp-servers/mcp-server:latest --region=asia-southeast1 --no-allow-unauthenticatedDeploying container to Cloud Run service [mcp-server]... Done....Service URL: https://mcp-server-214935060214.asia-southeast1.run.app

你的 MCP 服务器现在已经上线!

第 3 步:测试已部署的服务

由于我们部署的是私有服务,不能直接访问 URL。不过,gcloud 提供了一个安全本地代理,可为你处理认证。

图 69. 终端

> gcloud run services proxy mcp-server --region=asia-southeast1Proxying to Cloud Run service [mcp-server]...http://127.0.0.1:8080 proxies to https://mcp-server-bgynkccowq-as.a.run.app

这条命令创建了一条隧道:你在本机发往 http://127.0.0.1:8080 的请求会被安全转发到线上 Cloud Run 服务。

现在打开一个新终端,运行先前创建的 client.py

图 70. 终端

> python .\client.py[Tool(name='count_characters', ...)]24

成功!你的本地客户端脚本已经与全球可达、可扩展且安全的 MCP 服务器完成了通信。

第 4 步:清理资源

不再使用的资源最好及时清理,以免产生费用。

图 71. 终端

> gcloud run services delete mcp-server --region=asia-southeast1Service [mcp-server] will be deleted.Do you want to continue (Y/n)?  YDeleted service [mcp-server].

在虚拟机上手动部署

有时你需要比无服务器平台更高的掌控力。将应用部署到 VM 能让你完全控制操作系统、网络与已安装软件。代价是你需要自己负责运维与安全

通用做法:让 Python 服务器监听一个高位端口(如 8080 或 9000),并用成熟的 Web 服务器 Nginx 作为反向代理。Nginx 监听标准端口(80/443),再把流量转发给应用。

第 1 步:准备 VM 与应用

先在你喜欢的云商处开一台 VM(例如 GCP/AWS/DigitalOcean 上的 Ubuntu 24.04 LTS)。确保服务器能接收 HTTP 流量。SSH 登录后:

创建服务器目录、用 uv 配环境、安装 fastmcp。

图 72. VM 终端

$ mkdir mcp_server$ cd mcp_server$ wget -qO- https://astral.sh/uv/install.sh | sh$ source $HOME/.local/bin/env$ uv init$ uv add fastmcp

在 VM 上创建 server.py

图 73. server.py

from fastmcp import FastMCPmcp = FastMCP("My MCP Server")@mcp.tool()def count_characters(string: str) -> int:    return len(string)if __name__ == "__main__":    mcp.run(transport="http", host="0.0.0.0", port=9000)

这与 Cloud Run 的示例相同,只是端口换成 9000

第 2 步:让 MCP 服务器常驻运行

不能直接 python server.py 后关闭 SSH,因为进程会随会话终止。一个简单有效的后台运行工具是 screen

图 74. VM 终端

# 启动名为 'mcp' 的 screen 会话$ screen -S mcp# 在 screen 中进入项目并运行服务器$ cd mcp_server$ source .venv/bin/activate$ python server.py# 按 Ctrl+A 再 Ctrl+D 退出会话,进程仍在后台运行# 随时可用 `screen -r mcp` 重新连接

第 3 步:将 Nginx 配置为反向代理

安装 Nginx 并设置为开机自启。

图 75. VM 终端

$ sudo apt update$ sudo apt install nginx -y$ sudo systemctl enable nginx

为服务创建 Nginx 配置。假定域名是 app.example.com,希望在路径 /mcp-server/ 下提供 MCP 服务。

图 76. VM 终端

$ sudo vim /etc/nginx/sites-available/app.example.com.conf

加入以下配置:

图 77. /etc/nginx/sites-available/app.example.com.conf

server {        listen 80;        listen [::]:80;        server_name app.example.com;        location /mcp-server/ {                proxy_pass http://localhost:9000/;                proxy_set_header Host $host;                proxy_set_header X-Real-IP $remote_addr;                proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;                proxy_set_header X-Forwarded-Proto $scheme;        }        root /var/www/app.example.com;        index index.html;        location / {                try_files $uri $uri/ =404;        }}

启用该配置、创建测试页并重启 Nginx:

图 78. VM 终端

$ sudo ln -s /etc/nginx/sites-available/app.example.com.conf /etc/nginx/sites-enabled/$ sudo mkdir -p /var/www/app.example.com$ echo "Nginx is working!" | sudo tee /var/www/app.example.com/index.html$ sudo systemctl restart nginx

第 4 步:测试

在本地机器上,先让你的电脑知道 app.example.com 指向 VM 的公网 IP。编辑 hosts 文件:

添加一行:YOUR_VM_IP_ADDRESS app.example.com

在 Windows 上可通过下列命令刷新 DNS 解析缓存:

图 79. 刷新 DNS 缓存

> ipconfig /flushdns

在本地创建一个指向新 Nginx 代理 URL 的客户端脚本:

图 80. local_vm_client.py

import asynciofrom fastmcp import Clientfrom fastmcp.client.transports import (    StreamableHttpTransport,)client = Client(transport=StreamableHttpTransport("http://app.example.com/mcp-server/mcp/"))async def main():    async with client:                tools = await client.list_tools()        print(tools)                output = await client.call_tool("count_characters", {"string": "Strawberry is delicious!"})        extracted_text = output.content[0].text        print(extracted_text)asyncio.run(main())

运行该客户端。它会连接到 app.example.com,由 Nginx 接收请求并转发到正在运行的 Python 进程,你将获得返回结果。至此,你已在传统 VM 上成功部署并对外暴露你的 MCP 服务器。

关键要点(Key Takeaways)

本章带你完成了从本地原型到生产级服务的关键旅程,为你的 MCP 工具箱新增了两项至关重要的能力。

至此,你已具备构建、加固、部署健壮的、面向真实场景的 MCP 应用的完整能力。

下一章我们将探索高级服务器架构:你已经构建了独立的 MCP 服务器,但如果要把 MCP 能力集成进现有 Web 应用呢?我们将深入讲解如何将 MCP 与标准 Python Web 框架(如 Starlette)结合,把 MCP 服务器挂载为更大 API 的一部分,并构建强大的 ASGI 中间件。你将学会让 MCP 成为你既有 Web 生态中的“一等公民”。

Fish AI Reader

Fish AI Reader

AI辅助创作,多种专业模板,深度分析,高质量内容生成。从观点提取到深度思考,FishAI为您提供全方位的创作支持。新版本引入自定义参数,让您的创作更加个性化和精准。

FishAI

FishAI

鱼阅,AI 时代的下一个智能信息助手,助你摆脱信息焦虑

联系邮箱 441953276@qq.com

相关标签

MCP 服务器安全 JWT Docker Cloud Run Nginx 部署
相关文章