跳转至

为你的mcp server 添加认证功能

使用 python 开发mcp server 非常简单，但是官方文档对于在sse 或者 streamable 服务里如何添加认证写的很不清晰，我们可以非常快速的开发一个mcp server 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
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 参数。

SSE 类型

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
from mcp.server.fastmcp import FastMCP
import datetime
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response
from starlette.requests import Request
import uvicorn
import anyio


mcp = FastMCP("auth-mcp-server")


class VerifyHeaderMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        if request.url.path == "/messages/":
            authorization = request.headers.get("authorization")
            if not authorization:
                return Response("Unauthorized", status_code=401)
            else:
                # 做一些校验工作
                pass
        return await call_next(request)


@mcp.tool(name="获取当前时间", description="获取当前时间")
async def get_current_time() -> str:
    return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")


app = mcp.sse_app()
app.add_middleware(VerifyHeaderMiddleware)

async def run():
    config = uvicorn.Config(
            app,
            host="0.0.0.0",
            port=8000,
            log_level="info",
        )
    server = uvicorn.Server(config)
    await server.serve()

if __name__ == "__main__":
    anyio.run(run)

定义认证中间件

1
2
3
class VerifyHeaderMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
    ...

通过这个中间件，我们可以拿到 request 对象，这个对象包含请求的各种信息，其中 request.headers 为请求的headers 信息。通过 

1
authorization = request.headers.get("authorization")

拿到认证信息以后，在做一些自定义的处理，如果认证不通过的话，那么就返回401

自定义app

之前是直接使用

1
mcp.run(transport="sse")

来启动mcp server， 查看源码，

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
async def run_sse_async(self, mount_path: str | None = None) -> None:
        """Run the server using SSE transport."""
        import uvicorn
        starlette_app = self.sse_app(mount_path)
        config = uvicorn.Config(
            starlette_app,
            host=self.settings.host,
            port=self.settings.port,
            log_level=self.settings.log_level.lower(),
        )
        server = uvicorn.Server(config)
        await server.serve()

创建了一个 Starlette 对象starlette_app，之后再创建 uvicorn.Server 创建一个 server 对象，最后启动服务。

这里我们不能直接使用 mcp.run(transport="sse") 来启动服务，这种方式是没有机会将上面的认证中间件添加进去， 需要自定义 Starlette app

1
2
app = mcp.sse_app()
app.add_middleware(VerifyHeaderMiddleware)

通过 app = mcp.sse_app() 拿到 Starlette 对象以后，就可以通过 app.add_middleware(VerifyHeaderMiddleware) 将中间件添加到app中。

启动服务

1
2
3
4
5
6
7
8
9
async def run():
    config = uvicorn.Config(
            app,
            host="0.0.0.0",
            port=8000,
            log_level="info",
        )
    server = uvicorn.Server(config)
    await server.serve()

最后，通过server = uvicorn.Server(config) 自定义server， 然后就可以正常启动带有认证中间件的 mcp server 了。

streamable 类型

streamable 和 sse 很类似，只是请求的路径不同，我们创建一个带有认证功能的streamable mcp server. 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
from mcp.server.fastmcp import FastMCP
import datetime
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response
from starlette.requests import Request
import uvicorn
import anyio

mcp = FastMCP("auth-mcp-server")

class VerifyHeaderMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        if request.url.path == "/mcp/":
            authorization = request.headers.get("authorization")
            if not authorization:
                return Response("Unauthorized", status_code=401)
            else:
                # 做一些校验工作
                pass
        return await call_next(request)


@mcp.tool(name="获取当前时间", description="获取当前时间")
async def get_current_time() -> str:
    return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")


app = mcp.streamable_http_app()

app.add_middleware(VerifyHeaderMiddleware)

async def run():
    config = uvicorn.Config(
            app,
            host="0.0.0.0",
            port=8000,
            log_level="info",
        )
    server = uvicorn.Server(config)
    await server.serve()

if __name__ == "__main__":
    anyio.run(run)

和 sse 就两点不同

  1. app 初始化不同： sse 通过 app = mcp.sse_app() 初始化， streamable 通过 app = mcp.streamable_http_app() 初始化。
  2. 路径判断不同： sse 的请求路径为 if request.url.path == "/messages/":, streamable 是 if request.url.path == "/mcp/": 。

其他的代码都相同。

验证

使用 cherry studio 做个测试

image.png

如果在请求头处没有配置 Authorization 参数，会提示错误

image.png

配置上请求头则可以正常的添加 mcp server

image.png