云计算百科
云计算领域专业知识百科平台

如何使用 Python 和 FastAPI 构建带认证的 MCP 服务器(含代码)

曾经只存在于科幻想象中的场景——AI与任何应用程序无缝对接,如今正逐步成为现实。就像API长期作为开发者与软件进行交互的接口一样,模型上下文协议(Model Context Protocol,MCP)正逐渐成为AI智能体以结构化、感知上下文的方式与应用程序交互的首选标准。诸如Anthropic(该协议的创立者)、OpenAI、谷歌等众多AI供应商都在广泛采用这一协议。

对于应用程序开发者和维护者而言,用户通过AI智能体而非直接与应用交互的时代已经悄然来临,支持这一转变的关键就在于搭建MCP服务器。本文将详细介绍如何使用Python和FastAPI构建一个带有认证功能的MCP服务器,确保应用程序与AI智能体交互的安全性和高效性。

一、搭建前的准备工作

在开始搭建MCP服务器之前,需要先安装必要的依赖库。其中,uvicorn是一个轻量级的ASGI(Asynchronous Server Gateway Interface)服务器,用于运行FastAPI应用;fastapi是构建API的现代、快速且高效的框架;fastapi-mcp则是专门用于在FastAPI应用中集成MCP功能的轻量级开源库。使用以下命令进行安装:

pip install uvicorn fastapi fastapi-mcp

二、创建FastAPI应用

如果尚未创建FastAPI应用,那么这是首要步骤。FastAPI之所以适合用于搭建与AI智能体交互的应用,主要有以下几个优势:

  • 易于集成

    能够轻松与外部工具、服务或工作流程进行连接,满足多样化的业务需求。

  • 便于自动化

    支持用户自动化任务,减少人工操作可能带来的错误,提高工作效率。

  • 设计灵活

    在不改变核心逻辑的前提下,能够支持诸如MCP或移动应用等新的交互模型,具备良好的扩展性。

  • 面向未来

    逻辑与接口的清晰分离,使其能轻松适应新的协议或突发的需求变化。

  • 下面是一个创建简单FastAPI服务器的示例代码,该服务器仅有一个端点:

    from fastapi import FastAPIapp = FastAPI()@app.get("/")async def root():    return {"message": "MCP is super cool"}

    使用uvicorn main:app –reload命令运行该应用,就可以通过http://127.0.0.1:8000访问它。不过,在正式使用前,还需要为其添加MCP服务器功能。

    三、添加MCP服务器

    借助fastapi-mcp库,只需在FastAPI代码中添加两行代码,就能将所有已定义的端点作为MCP工具暴露出来:

    from fastapi import FastAPIfrom fastapi_mcp import FastApiMCPapp = FastAPI()mcp = FastApiMCP(app)mcp.mount()

    这样,MCP服务器就可以通过http://127.0.0.1:8000/mcp访问。该地址可在任何支持SSE(Server-Sent Events)传输的AI智能体中进行配置。如果需要进行更高级的配置,比如自定义工具暴露或挂载选项,可以参考fastapi-mcp的官方文档。

    四、配置MCP服务器的认证

    为确保MCP服务器的安全性,需要配置认证机制。fastapi-mcp支持两种主要的认证方法:授权头认证和OAuth2认证,其中OAuth2认证是推荐的方式,因为它完全符合MCP规范2025 – 03 – 26。

    (一)通过授权头进行令牌认证

    这是一种快速保护MCP服务器的方法,只需在MCP配置中提供一个授权头即可。在MCP配置文件中可以这样设置:

    {    "mcpServers": {        "remote-example": {            "command": "npx",            "args": [                "mcp-remote",                "http://localhost:8000/mcp",                "–header",                "Authorization:${AUTH_HEADER}"            ]        },        "env": {            "AUTH_HEADER": "Bearer <your-token>"        }    }}

    同时,为了拒绝未经授权的客户端连接到MCP服务器,还需要在创建MCP时调整认证依赖:

    from fastapi import FastAPI, Dependsfrom fastapi.security import HTTPBearerfrom fastapi_mcp import FastApiMCP, AuthConfigtoken_auth_scheme = HTTPBearer()app = FastAPI()mcp = FastApiMCP(    app,    name="Protected MCP",    auth_config=AuthConfig(        dependencies=[Depends(token_auth_scheme)],    ))mcp.mount()

    这里有一个可用的模板,开发者可以根据实际需求进行修改和完善。

    (二)使用Auth0进行完整的OAuth流程(推荐)

    这种方法遵循最新的MCP规范,是配置MCP服务器认证的推荐方式。下面以使用Auth0为例进行详细说明:

  • 添加Auth0配置建议将Auth0配置信息存储在项目根目录下的.env文件中,需要设置的变量包括: AUTH0_DOMAIN=your-tenant.auth0.comAUTH0_AUDIENCE=https://your-tenant.auth0.com/api/v2/AUTH0_CLIENT_ID=your-client-idAUTH0_CLIENT_SECRET=your-client-secret 通过创建一个Settings类来加载这些配置信息(在本指南中,为了简化,直接将其作为全局变量引用):
  • from pydantic_settings import BaseSettingsclass Settings(BaseSettings):    auth0_domain: str      auth0_audience: str    auth0_client_id: str    auth0_client_secret: str    class Config:        env_file = ".env"settings = Settings()

    2、在Auth0仪表板中正确配置回调URL:回调URL用于在认证过程中接收Auth0返回的信息,确保配置正确无误,否则认证流程可能无法正常进行。

    3、获取并转换JWK公钥:为了与Auth0进行认证,需要获取JWK(JSON Web Key)公钥,并将其转换为PEM格式

    import jwtimport requestsfrom cryptography.hazmat.primitives import serializationdef _get_public_key(url: str): response = requests.get(url) jwks = response.json() return jwks["keys"][0]def _convert_key_to_pem(jwk_key) -> str: public_key = jwt.algorithms.RSAAlgorithm.from_jwk(jwk_key) pem = public_key.public_bytes( encoding=serialization.Encoding.PEM, format=serialization.PublicFormat.SubjectPublicKeyInfo, ) return pem.decode("utf-8")jwks_public_key_from_auth_0 = _get_public_key(f"https://{AUTH0_DOMAIN}/.well-known/jwks.json")pem_key = _convert_key_to_pem(jwks_public_key_from_auth_0)

    4、验证认证令牌

    创建一个验证函数,用于验证访问权限。这个函数不仅可以用于FastAPI中的特定私有端点,还将在后续配置MCP时用到。

    async def verify_auth(request: Request) -> dict[str, Any]: # 可能需要更多的验证逻辑 return jwt.decode(     token,     jwks_public_key_from_auth_0,     algorithms=["RS256", "HS256"],     audience=AUTH0_AUDIENCE,     issuer=f"https://{AUTH0_DOMAIN}/",     options={"verify_signature": True}, )

    5、使用OAuth保护MCP服务器

    完成上述步骤后,就可以配置OAuth来保护MCP服务器了。 ​​​​​​​

    from fastapi import Dependsfrom fastapi_mcp import FastApiMCP, AuthConfigmcp = FastApiMCP(    app,    name="MCP With Auth0",    auth_config=AuthConfig(        issuer=f"https://{AUTH0_DOMAIN}/",        authorize_url=f"https://{AUTH0_DOMAIN}/authorize",        oauth_metadata_url=f"https://{AUTH0_DOMAIN}/.well-known/openid-configuration",        audience=AUTH0_AUDIENCE,        client_id=AUTH0_CLIENT_ID,        client_secret=AUTH0_CLIENT_SECRET,        dependencies=[Depends(verify_auth)],         setup_proxies=True,    ))mcp.mount()

    这样,MCP服务器就通过Auth0实现了全面的安全保护。如果计划在生产环境中使用,强烈建议参考项目示例部分的完整代码示例,以确保安全性和稳定性。

    当基于FastAPI的MCP服务器搭建完成并投入运行后,就可以开始构建更强大、能感知上下文的应用程序,实现与AI智能体的无缝交互。这仅仅是一个开始,未来还有许多令人期待的功能。例如,基于新的MCP规范的可流式HTTP传输,以及自定义MCP工具,这些工具允许开发者添加独立于FastAPI端点的工具和资源,进一步拓展应用程序的功能边界。

    同时,这个开源项目也欢迎广大开发者参与贡献和改进。无论是报告问题、提出新功能建议,还是提交拉取请求,每一份贡献都将有助于塑造这个库的未来,提升其功能和性能。

    通过以上步骤,开发者可以成功在Python中使用FastAPI构建一个带有认证功能的MCP服务器。在实际应用中,还需要根据具体的业务需求和安全要求进行进一步的优化和扩展,以充分发挥MCP在AI与应用程序交互中的优势,为用户提供更加智能、高效的服务体验。

    code(https://github.com/tadata-org/fastapi_mcp/blob/main/examples/09_auth_example_auth0.py)

    赞(0)
    未经允许不得转载:网硕互联帮助中心 » 如何使用 Python 和 FastAPI 构建带认证的 MCP 服务器(含代码)
    分享到: 更多 (0)

    评论 抢沙发

    评论前必须登录!