2025年12月29日月曜日

MCPサーバの戻り値

特に戻り値の設定の仕方についての検討を行いました。

MCP 2025-06-18 で追加された structured tool output を試す

MCP の Structured tool output を試してみる

MCP サーバーでの outputSchema の注意点

了解です、孝幸さん。Python で MCP サーバを書く場合、**戻り値は「辞書（dict）で content を返す」**という形になります。
Node 版と同じ構造ですが、Python ではよりシンプルに書けます。


ここからは、実際に動く最小の Python 実装例を使って、戻り値の書き方をはっきり示します。

🐍 Python MCP サーバの戻り値はこう書く

MCP のツールは、Python の関数が dict を返すだけで OK です。

基本形：

return {
    "content": [
        {
            "type": "text",
            "text": "ここに返したい文字列"
        }
    ]
}

📘 最小の MCP サーバ（Python）

from mcp.server import Server
from mcp.types import TextContent

server = Server(name="example-python-server", version="1.0.0")

@server.tool()
def hello(name: str):
    return {
        "content": [
            TextContent(
                type="text",
                text=f"こんにちは、{name} さん！"
            )
        ]
    }

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

ポイント：

  • @server.tool() でツールを定義
  • 戻り値は content: [TextContent(...)]
  • TextContent を使うと型安全で便利

🧪 もう少し複雑な戻り値（複数の content）

return {
    "content": [
        {"type": "text", "text": "処理が完了しました"},
        {"type": "text", "text": "追加情報: OK"}
    ]
}

❗ エラーを返したい場合

Python の MCP SDK では、例外を投げるだけで JSON-RPC の error に変換されます。

from mcp.server import Server, ToolError

@server.tool()
def divide(a: float, b: float):
    if b == 0:
        raise ToolError("ゼロ除算はできません")
    return {
        "content": [
            {"type": "text", "text": str(a / b)}
        ]
    }

🎯 まとめ

Python MCP サーバの戻り値は：

  • dict を返すだけでよい
  • content は 配列
  • 各要素は {"type": "text", "text": "..."}
  • エラーは ToolError を投げる

必要なら：

  • Python でのストリーミング返却
  • JSON 以外（画像など）の返し方
  • FastAPI や asyncio との組み合わせ

なども詳しく書けますよ。どこまで作りたいか教えてください。

Python で MCP のストリーミング返却（partial response） を行う方法を、実際に動くコード付きで分かりやすくまとめます。

結論から言うと：

🎯 Python MCP のストリーミング返却は「yield を使う」

MCP の Python SDK では、ツール関数を async generator（非同期ジェネレータ） にすると、
yield した内容が 部分レスポンス（partial response） としてクライアントに送られます。

🧪 最小のストリーミング例（Python）

from mcp.server import Server
from mcp.types import TextContent
import asyncio

server = Server(name="stream-example", version="1.0.0")

@server.tool()
async def stream_test():
    # 1つ目の部分レスポンス
    yield {
        "content": [
            TextContent(type="text", text="処理を開始します…")
        ]
    }

    await asyncio.sleep(1)

    # 2つ目の部分レスポンス
    yield {
        "content": [
            TextContent(type="text", text="まだ処理中です…")
        ]
    }

    await asyncio.sleep(1)

    # 最終レスポンス（return）
    return {
        "content": [
            TextContent(type="text", text="完了しました！")
        ]
    }

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

🧩 仕組みのポイント

✔ yield → partial response

クライアント（例：Copilot）は、yield された内容を 逐次受け取る。

✔ return → final response

最後に return した内容が 最終レスポンス になる。

✔ 非同期ジェネレータである必要がある

async def + yield の組み合わせ。

📡 クライアント側の受け取りイメージ

クライアントはこんな順番で受け取る：

  1. 「処理を開始します…」
  2. 「まだ処理中です…」
  3. 「完了しました！」

Copilot や VS Code の MCP クライアントは、これをリアルタイムに表示できる。

🔍 もう少し実用的な例：長い処理の進捗を送る

@server.tool()
async def long_task():
    for i in range(5):
        yield {
            "content": [
                TextContent(type="text", text=f"進捗: {i+1}/5")
            ]
        }
        await asyncio.sleep(0.5)

    return {
        "content": [
            TextContent(type="text", text="すべて完了しました！")
        ]
    }

🎁 まとめ

Python MCP のストリーミング返却は：

  • async def でツールを定義
  • yield で部分レスポンスを送る
  • return で最終レスポンスを返す
  • content は通常と同じ構造（TextContent など）

必要なら：

  • ストリーミングで画像を返す方法
  • 大規模処理の進捗バーを送る方法
  • FastAPI や asyncio タスクとの連携

なども作れますよ。


Q.たとえば、get_constraintsに対して、どのように結果を返せばよいですか？スケジュールナース内の定義をどのように教えればよいですか？

ここは誤解しやすいポイントなので、丁寧に整理しますね。
結論：MCP サーバは “Python のデータ構造をそのまま返す” のではなく、MCP の content 形式に変換して返す必要があります。
つまり、get_constraints がどんな構造の制約を返したいかを あなたが定義し、それを content に載せて返す という形になります。

🎯 まず前提：MCP は「任意の構造体」を返せる

MCP のツールは、次のような構造で返します：

{
  "content": [
    {
      "type": "text",
      "text": "..."
    }
  ]
}

つまり、返したいデータは text（または json）として content に入れる必要があります。

🧩 では、get_constraints の戻り値はどうする？

あなたの「スケジュールナース（Schedule Nurse）」の制約定義が、例えば Python でこうだとします：

constraints = {
    "max_shifts_per_day": 2,
    "max_consecutive_days": 5,
    "required_rest_hours": 8,
    "forbidden_pairs": [
        ["Aさん", "Bさん"],
        ["Cさん", "Dさん"]
    ]
}

これを MCP の返却形式にすると：

import json
from mcp.types import TextContent

@server.tool()
def get_constraints():
    constraints = {
        "max_shifts_per_day": 2,
        "max_consecutive_days": 5,
        "required_rest_hours": 8,
        "forbidden_pairs": [
            ["Aさん", "Bさん"],
            ["Cさん", "Dさん"]
        ]
    }

    return {
        "content": [
            TextContent(
                type="text",
                text=json.dumps(constraints, ensure_ascii=False, indent=2)
            )
        ]
    }

📦 JSON で返すのがベスト

MCP は type="json" もサポートしています。
Schedule Nurse のように構造化データを返すなら、こちらの方が自然です。

from mcp.types import JsonContent

@server.tool()
def get_constraints():
    constraints = {
        "max_shifts_per_day": 2,
        "max_consecutive_days": 5,
        "required_rest_hours": 8,
        "forbidden_pairs": [
            ["Aさん", "Bさん"],
            ["Cさん", "Dさん"]
        ]
    }

    return {
        "content": [
            JsonContent(
                type="json",
                json=constraints
            )
        ]
    }

🧠 「スケジュールナース内の定義をどう教える？」の答え

あなたが Schedule Nurse 内で持っている制約定義（Python の dict、クラス、DB など）は：

  • MCP サーバ内で読み込む
  • 必要な形に整形する
  • MCP の content として返す

という流れになります。

つまり、Schedule Nurse の内部構造をそのまま返すのではなく、MCP クライアントが理解できる JSON 形式に変換して返すのが正解です。

🎁 さらに実用的な例：制約をストリーミングで返す

制約が大量なら、ストリーミングも可能です。

@server.tool()
async def get_constraints():
    yield {
        "content": [
            TextContent(type="text", text="基本制約を読み込み中…")
        ]
    }

    await asyncio.sleep(0.5)

    constraints = load_constraints_from_schedule_nurse()

    yield {
        "content": [
            JsonContent(type="json", json=constraints)
        ]
    }

    return {
        "content": [
            TextContent(type="text", text="完了しました")
        ]
    }

🎯 まとめ

  • MCP の返却形式は content 配列
  • get_constraints は JSON を返すのが最適
  • Schedule Nurse の内部定義は Python で読み込み → JSON に変換 → content に載せる
  • ストリーミング返却も可能

必要なら：

  • あなたの Schedule Nurse の制約定義（Python コード）を MCP 用に最適化する方法
  • 制約の種類ごとにツールを分ける設計
  • 制約の更新 API（set_constraints）の作り方

なども一緒に作れますよ。



投稿者 時刻:

0 件のコメント:

コメントを投稿

登録: コメントの投稿 (Atom)