为你的mcp server 添加认证功能
使用 python 开发mcp server 非常简单,但是官方文档对于在sse 或者 streamable 服务里如何添加认证写的很不清晰,我们可以非常快速的开发一个mcp server
from mcp.server.fastmcp import FastMCP
import datetime
mcp = FastMCP("simple-mcp-server")
@mcp.tool(name="获取当前时间", description="获取当前时间")
async def get_current_time() -> str:
return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
def run():
mcp.run(transport="sse")
if __name__ == "__main__":
run()
以上是一个获取当前时间的mcp server ,很简单,只作为演示,但是如果我们不想让所有人都可以访问,需要添加一些认证信息,目前来讲,由于FastMCP 封装的太高级了,不太好添加,官方有提供基于 OAuth 的认证方式,但是有点复杂,我还没有测试成功过,本文介绍一种简单的在header 中添加认证信息。
这里分别演示 SSE 和 streamable 两种类型如何改造,原理差不太多, 通过创建一个自定义的中间件,在请求处理之前获取一下header 中的 Authorization 参数。