MCP のトランスポート層に認証機能を付ける - フォトシンスエンジニアブログ

どうも daikw - Qiita です。

先週読んだ MCP の仕様で気になった部分があったので、実装してみました。

背景

MCP の仕様では、トランスポート層に認証機能を付けることを推奨しています。

https://modelcontextprotocol.io/docs/concepts/transports#authentication-and-authorization

Authentication and Authorization - Implement proper authentication mechanisms - Validate client credentials - Use secure token handling - Implement authorization checks

ただ、 modelcontextprotocol/python-sdk を読んでいると、まだ実装はされていないようです。

どうやらカスタムトランスポートの実装を期待しているように見えたので、その方向で実装してみます。

調査

現在までのほとんどの MCP サーバを実装した記事は、いずれも stdio Transport を使用していて、 SSE Transport のサンプルがほとんどありませんでした。

さらに言えば、現状 Claude Desktop 自体も stdio Transport しかサポートしていません。

https://github.com/orgs/modelcontextprotocol/discussions/16#discussion-7565289

modelcontextprotocol/python-sdk が mcp という名前の Python パッケージとして公開されており、いくつかのサンプル実装が提供されています。

uvx create-mcp-server で生成されるもの
python-sdk/examples at main · modelcontextprotocol/python-sdk のサンプル

1 は stdio Transport の実装を提供します。 2 には SSE Transport の実装サンプルを提供していますが、これはそのままでは動きませんでした。

https://github.com/modelcontextprotocol/python-sdk/tree/aaf32b530738ff79ba607c2884374243350f521c/examples/servers/simple-tool

┬─[daiki@mac42:~/g/g/m/p/e/s/simple-tool]─[20:16:08]─[G:main=]
╰─>$ uv run mcp-simple-tool --transport sse --port 8000
INFO:     Started server process [67841]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
INFO:     127.0.0.1:57607 - "GET /sse HTTP/1.1" 200 OK
INFO:     127.0.0.1:57608 - "POST /messages?session_id=4ec879685e0e482e84f008362094dded HTTP/1.1" 202 Accepted
ERROR:    Exception in ASGI application
Traceback (most recent call last):
  File "/Users/daikiwatanabe/ghq/github.com/modelcontextprotocol/python-sdk/.venv/lib/python3.10/site-packages/uvicorn/protocols/http/h11_impl.py", line 406, in run_asgi
    result = await app(  # type: ignore[func-returns-value]
  File "/Users/daikiwatanabe/ghq/github.com/modelcontextprotocol/python-sdk/.venv/lib/python3.10/site-packages/uvicorn/middleware/proxy_headers.py", line 60, in __call__
    return await self.app(scope, receive, send)
  File "/Users/daikiwatanabe/ghq/github.com/modelcontextprotocol/python-sdk/.venv/lib/python3.10/site-packages/starlette/applications.py", line 113, in __call__
    await self.middleware_stack(scope, receive, send)
  File "/Users/daikiwatanabe/ghq/github.com/modelcontextprotocol/python-sdk/.venv/lib/python3.10/site-packages/starlette/middleware/errors.py", line 187, in __call__
    raise exc
  File "/Users/daikiwatanabe/ghq/github.com/modelcontextprotocol/python-sdk/.venv/lib/python3.10/site-packages/starlette/middleware/errors.py", line 165, in __call__
    await self.app(scope, receive, _send)
  File "/Users/daikiwatanabe/ghq/github.com/modelcontextprotocol/python-sdk/.venv/lib/python3.10/site-packages/starlette/middleware/exceptions.py", line 62, in __call__
    await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send)
  File "/Users/daikiwatanabe/ghq/github.com/modelcontextprotocol/python-sdk/.venv/lib/python3.10/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app
    raise exc
  File "/Users/daikiwatanabe/ghq/github.com/modelcontextprotocol/python-sdk/.venv/lib/python3.10/site-packages/starlette/_exception_handler.py", line 42, in wrapped_app
    await app(scope, receive, sender)
  File "/Users/daikiwatanabe/ghq/github.com/modelcontextprotocol/python-sdk/.venv/lib/python3.10/site-packages/starlette/routing.py", line 715, in __call__
    await self.middleware_stack(scope, receive, send)
  File "/Users/daikiwatanabe/ghq/github.com/modelcontextprotocol/python-sdk/.venv/lib/python3.10/site-packages/starlette/routing.py", line 735, in app
    await route.handle(scope, receive, send)
  File "/Users/daikiwatanabe/ghq/github.com/modelcontextprotocol/python-sdk/.venv/lib/python3.10/site-packages/starlette/routing.py", line 288, in handle
    await self.app(scope, receive, send)
  File "/Users/daikiwatanabe/ghq/github.com/modelcontextprotocol/python-sdk/.venv/lib/python3.10/site-packages/starlette/routing.py", line 76, in app
    await wrap_app_handling_exceptions(app, request)(scope, receive, send)
  File "/Users/daikiwatanabe/ghq/github.com/modelcontextprotocol/python-sdk/.venv/lib/python3.10/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app
    raise exc
  File "/Users/daikiwatanabe/ghq/github.com/modelcontextprotocol/python-sdk/.venv/lib/python3.10/site-packages/starlette/_exception_handler.py", line 42, in wrapped_app
    await app(scope, receive, sender)
  File "/Users/daikiwatanabe/ghq/github.com/modelcontextprotocol/python-sdk/.venv/lib/python3.10/site-packages/starlette/routing.py", line 74, in app
    await response(scope, receive, send)
TypeError: 'NoneType' object is not callable
...

実装

先ほどの ASGI サーバの振る舞いを満たさない実装は、 starlette.Route を starlette.Mount に差し替えれば動作します。 https://github.com/modelcontextprotocol/python-sdk/pull/83#pullrequestreview-2486695453

       starlette_app = Starlette(
           debug=True,
           routes=[
               Route("/sse", endpoint=handle_sse),
[+]            Mount("/messages/", app=sse.handle_post_message),
[-]            Route("/messages", endpoint=handle_messages, methods=["POST"]),
           ],
       )

その上で mcp.server.sse.SseServerTransport をラップし、 Basic 認証機能をつけます。

import base64
from contextlib import asynccontextmanager

from starlette.types import Scope, Receive, Send
from starlette.exceptions import HTTPException

from mcp.server.sse import SseServerTransport

class BasicAuthTransport(SseServerTransport):
    """
    Example basic auth implementation of SSE server transport.
    """
    def __init__(self, endpoint: str, username: str, password: str):
        super().__init__(endpoint)
        self.expected_header = b"Basic " + base64.b64encode(f"{username}:{password}".encode())

    @asynccontextmanager
    async def connect_sse(self, scope: Scope, receive: Receive, send: Send):
        auth_header = dict(scope["headers"]).get(b'authorization', b'')
        if auth_header != self.expected_header:
            raise HTTPException(status_code=401, detail="Unauthorized")
        async with super().connect_sse(scope, receive, send) as streams:
            yield streams

クライアント側にも Authorization ヘッダをつけます。

import asyncio
import base64
import click
import os
from mcp.client.session import ClientSession

from mcp.client.sse import sse_client

username = os.environ.get("USERNAME")
password = os.environ.get("PASSWORD")

async def __main(endpoint: str):
    async with sse_client(
        url=endpoint,
        headers={"Authorization": "Basic " + base64.b64encode(f"{username}:{password}".encode()).decode("utf-8")},
    ) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # List available tools
            tools = await session.list_tools()
            print(tools)

            # Call the execute_command tool
            result = await session.call_tool("execute_command", {"command": "id", "args": ["-un"]})
            print("-" * 100)
            for content in result.content:
                print(content.text)


@click.command()
@click.option("--endpoint", default="http://localhost:8000/sse", help="URL of the SSE endpoint")
def main(endpoint: str):
    asyncio.run(__main(endpoint))

これでようやく動きます。

今回の実装は daikw/mcp-transport-auth に公開しました。 uv で動かせます。

git clone https://github.com/daikw/mcp-transport-auth.git
cd mcp-transport-auth

# terminal 1: server
USERNAME=admin PASSWORD=pass uv run mcp-transport-auth --port 8000

# terminal 2: client
USERNAME=admin PASSWORD=pass uv run mcp-transport-auth-client

サーバ側

❯ USERNAME=admin PASSWORD=pass uv run mcp-transport-auth --port 8000
INFO:     Started server process [70792]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
INFO:     127.0.0.1:57863 - "GET /sse HTTP/1.1" 200 OK
INFO:     127.0.0.1:57865 - "POST /messages/?session_id=f6640184a56e43fb93f6b697bfd936b7 HTTP/1.1" 202 Accepted
INFO:     127.0.0.1:57865 - "POST /messages/?session_id=f6640184a56e43fb93f6b697bfd936b7 HTTP/1.1" 202 Accepted
INFO:     127.0.0.1:57865 - "POST /messages/?session_id=f6640184a56e43fb93f6b697bfd936b7 HTTP/1.1" 202 Accepted
INFO:mcp.server:Processing request of type ListToolsRequest
INFO:     127.0.0.1:57865 - "POST /messages/?session_id=f6640184a56e43fb93f6b697bfd936b7 HTTP/1.1" 202 Accepted
INFO:mcp.server:Processing request of type CallToolRequest
INFO:mcp_transport_auth.server:Executing command: execute_command {'command': 'id', 'args': ['-un']}
INFO:mcp_transport_auth.server:Command output: daikiwatanabe

クライアント側

❯ USERNAME=admin PASSWORD=pass uv run mcp-transport-auth-client
nextCursor=None tools=[Tool(name='execute_command', description='Executes a command and returns its output', inputSchema={'type': 'object', 'required': ['command', 'args'], 'properties': {'command': {'type': 'string', 'description': 'Command to execute'}, 'args': {'type': 'array', 'items': {'type': 'string'}, 'description': 'Arguments to pass to the command'}}})]
----------------------------------------------------------------------------------------------------
daikiwatanabe