🦔
MCPのPython SDK + FastAPIでMCPサーバーを起動する

2025/05/02に公開
Python
FastAPI
Model Context Protocol
tech
最近話題のMCP(Model-Context-Protocol)を学習していて、試しにMCPサーバーを作ってみました。

 やりたかったことstdioモードで起動する
sseモードで起動する、その際にFastAPIやFlask等のWebフレームワークを使用したい
FastAPI-MCPというライブラリもあるみたいですが、これはFastAPIで構築したエンドポイントをMCPに変換してくれるツールの模様。MCPサーバーの起動方法を任意に選択できるというわけではなさそうなので、今回は使用しません。学習目的も兼ねているので、MCPのPythonSDKを直接触ってみたかったのもあります。

 stdioで動かしてみる
 ソースコードサンプルを見ながらテキトーに作ってみる。検証に使用したツールはページの最後に載せておきます。
app.py
from mcp.server.fastmcp import FastMCP
from services.weather import WeatherService, WeatherServiceError


mcp_server = FastMCP("FastMCP")

@mcp_server.tool()
async def get_forecast(latitude: float, longitude: float) -> list[Any] | str:
    """Get weather forecast for a location.

    Args:
        latitude: Latitude of the location
        longitude: Longitude of the location
    """

    weather_service = WeatherService()

    try:
        return await weather_service.get_forecast(latitude, longitude)
    except WeatherServiceError as e:
        logger.error(f"Error fetching forecast: {e}")
        return "Cannot fetch forecast data."

if __name__ == "__main__":
    logger.info("FastMCP server started")
    mcp_server.run()
toolデコレータを設定した関数をtoolに追加してくれるみたいですね。

 実際に動かしてみる検証にはGitHub Copilotを使用しました。

 VS Codeの設定まずはVS Codeの設定から。User settings JSONを開いて以下の設定を追加。pathは適宜修正してください。
    "mcp": {
        "servers": {
            "sample-stdio-mcp-server": {
                "command": "/path/to/python",
                "args": [
                    "/path/to/app.py",
                ]
            }
        }
    },
    "chat.mcp.discovery.enabled": true,


 Copilot君に聞いてみたデフォルトではAskが選択されているはずなので、Agentに変更するのをお忘れなく。(私はこれでハマりました😄)
こんな感じで実行するかを確認されるのでContinueを押下。
応答が返ってきました！

 sseで動かしてみる
 検討した内容https://github.com/jlowin/fastmcp/blob/main/src/fastmcp/server/server.py
まずはFastMCPクラスのソースを見て、run_sse_asyncでMCPサーバーが起動することを確認。内部でsse_app関数を呼んでStarletteインスタンスを生成しており、外部から設定するのはできなそう。ただ、Starletteを継承しているFastAPIの方が親和性が高そうなので今回はこちらを採用。
sse_app関数を調べてみたところ、以下2点を設定すれば良さそう。
sseエンドポイント: SseServerTransportのconnect_sseでSSE接続を確立し、mcpサーバーを起動する
messageエンドポイント: SseServerTransportのhandle_post_messageを呼び出せば良さそう
sse_app関数で定義しているhandle_sseを使いたいけど、関数内で定義されているので直接は呼び出せない。似たような関数を独自に作成するにしてもプライベート定義している_mcp_serverを参照するのは抵抗があったため、FastMCPを継承したクラスを作ることに。

 実装from fastapi import FastAPI

from mcp.server import FastMCP
from mcp.server.sse import SseServerTransport
from starlette.requests import Request


class FastMCPApi(FastMCP):
    def __init__(self, app: FastAPI, name="FastMCPApi", mount_path: str = "/mcp"):
        super().__init__(name)

        self.app = app
        base_path = app.root_path
        self.mount_path = mount_path
        self.message_path = f"{base_path}{mount_path}/messages"
        self.__setup()

    def __setup(self):
        sse = SseServerTransport(self.message_path)

        @self.app.get(self.mount_path, operation_id="mcp_connection")
        async def handle_sse(request: Request) -> None:
            scope = request.scope
            recieve = request.receive
            send = request._send  # type: ignore[reportPrivateUsage]
            async with sse.connect_sse(scope, recieve, send) as (reader, writer):
                await self._mcp_server.run(
                    reader,
                    writer,
                    self._mcp_server.create_initialization_options(),
                )

        @self.app.post(self.message_path, operation_id="mcp_message")
        async def handle_mcp_message(request: Request):
            scope = request.scope
            recieve = request.receive
            send = request._send  # type: ignore[reportPrivateUsage]
            return await sse.handle_post_message(scope, recieve, send)

    def run(self, host="127.0.0.1", port=8000):
        from uvicorn import run

        run(self.app, host=host, port=port, log_level="info")
sse版のエントリポイントはstdio版とわけようと思ったので、tool登録処理を別ファイルに外だしして共通利用できるように修正。
from typing import Any
from mcp.server.fastmcp import FastMCP

from logging import getLogger

logger = getLogger(__name__)

from services.weather import WeatherService, WeatherServiceError


def register_tool(mcp_server: FastMCP):
    @mcp_server.tool()
    async def get_forecast(latitude: float, longitude: float) -> list[Any] | str:
        """Get weather forecast for a location.

        Args:
            latitude: Latitude of the location
            longitude: Longitude of the location
        """

        weather_service = WeatherService()

        try:
            return await weather_service.get_forecast(latitude, longitude)
        except WeatherServiceError as e:
            logger.error(f"Error fetching forecast: {e}")
            return "Cannot fetch forecast data."
server.pyとして起動処理を定義
import logging


from fastapi import FastAPI

from mcp_def import register_tool
from shared.mcp import FastMCPApi

# ログ設定
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Flaskアプリケーションの作成
app = FastAPI(title=__name__)

mcp_server = FastMCPApi(app, name="FastMCPApi", mount_path="/mcp")
register_tool(mcp_server)

# Flaskアプリケーションの起動
if __name__ == "__main__":
    logger.info("Starting Flask MCP server...")
    mcp_server.run()

 実際に動かしてみる
 サーバーの起動とCopilotの設定まずはサーバーの起動
python server.py
続いてVS Codeの設定から。User settings JSONに以下を追記
            "sample-sse-mcp-server": {
                "type": "sse",
                "url": "http://localhost:8000/mcp"
            }
sample-stdio-mcp-serverの方は停止して、sample-sse-mcp-serverを起動。以下のログが出たので、copilot側からアクセスができた模様。
INFO:mcp.server.lowlevel.server:Processing request of type ListToolsRequest

 再度Copilot君に聞いてみる設定は完了したので、先ほどと同様にCopilotに質問。
結果が返ってきました！ちゃんとFastAPIで起動したエンドポイントにリクエストが届いています。

 振り返り今回はMCP Python SDKを学びながらFastAPIでMCPサーバーを起動する方法を実装しました。上記検証後にアクセスログを出力するMiddlewareを追加してみましたが、こちらも正しく動作しました。慣れ親しんだFastAPIのフレームワークを活用しながらMCPサーバーを開発することができそうです。
app = FastAPI(title=__name__)

app.add_middleware(
    AccessLoggingMiddleware,
)
MCPサーバーを開発していて外部に公開したい、その際にFastAPIのフレームワークを活用したいという方の助けになれば幸いです。ただ、私もまだまだ学習している段階なので、もっといい方法があればぜひコメントをお願いします！

 サンプルで使用したツール最後に今回使用したサンプルのツールを載せておきます。
無料で使用可能なNotion Weather Serviceを使用して天気予報を取得するツールを作成して動作確認を行いました。
import httpx
from typing import Any

NWS_API_BASE = "https://api.weather.gov"


async def request_to_nws(url: str) -> dict[str, Any]:
    """Make a request to the NWS API with proper error handling."""
    headers = {"User-Agent": "weather-app/1.0", "Accept": "application/geo+json"}
    async with httpx.AsyncClient() as client:
        response = await client.get(url, headers=headers, timeout=10.0)
        response.raise_for_status()
        return response.json()


class WeatherServiceError(Exception):
    """Custom exception for WeatherService errors."""

    pass


class PointsError(WeatherServiceError):
    """Exception raised for errors in the points request."""

    pass


class ForecastError(WeatherServiceError):
    """Exception raised for errors in the forecast request."""

    pass


class WeatherService:
    """Service to interact with the National Weather Service API."""

    def __init__(
        self,
        api_base: str = NWS_API_BASE,
        num_of_periods: int = 5,
    ):
        self.nws_api_base = api_base
        self.num_of_periods = num_of_periods

    async def __get_points(self, latitude: float, longitude: float) -> dict[str, Any]:
        """Get the forecast grid endpoint for a given latitude and longitude."""
        try:
            points_url = f"{self.nws_api_base}/points/{latitude},{longitude}"
            points_data = await request_to_nws(points_url)

            if not points_data:
                raise PointsError("Unable to fetch points data.")

            return points_data
        except httpx.RequestError as e:
            raise PointsError() from e

    async def __get_forecast(self, forecast_url: str) -> dict[str, Any]:
        """Get the forecast data from the forecast URL."""
        try:
            forecast_data = await request_to_nws(forecast_url)

            if not forecast_data:
                raise ForecastError("Unable to fetch forecast data.")

            return forecast_data
        except httpx.RequestError as e:
            raise ForecastError() from e

    async def get_forecast(self, latitude: float, longitude: float) -> str:
        """Get weather forecast for a location.

        Args:
            latitude: Latitude of the location
            longitude: Longitude of the location
        """
        points_data = await self.__get_points(latitude, longitude)
        forecast_data = await self.__get_forecast(points_data["properties"]["forecast"])

        periods = forecast_data["properties"]["periods"]
        return periods[: self.num_of_periods]
やりたかったこと

stdioで動かしてみる

ソースコード

実際に動かしてみる

VS Codeの設定

Copilot君に聞いてみた

sseで動かしてみる

検討した内容

実装

実際に動かしてみる

サーバーの起動とCopilotの設定

再度Copilot君に聞いてみる

振り返り

サンプルで使用したツール

Discussion
