MCP サーバーから Claude を逆に呼び出す Sampling 機能の実装 ― createMessage コールバックと権限承認の3つのハマりどころ【2026】

はじめに / 対象と前提

MCP(Model Context Protocol)には、サーバー側からクライアント(ホストアプリ)に対して「LLM に代わりに推論させてほしい」とリクエストできる Sampling という仕組みがある。普段 MCP サーバーを作るときは「ツールを生やして呼ばれるのを待つ」側に回ることが多いが、Sampling を使うとサーバー側から sampling/createMessage を叩いて、ホストが保持している LLM(Claude)にメッセージを投げ、応答を受け取れる。

この記事は以下を前提にしている。

• 自作 MCP サーバー(Python の mcp SDK)を最低限触ったことがある
• Claude Code など MCP ホスト経由でサーバーを動かす前提の話(Sampling はホストの実装依存が大きい)
• 環境:mcp Python SDK 1.x 系、Claude Code v1.x、Python 3.13

TL;DR を先に書くと、Sampling は「サーバーがクライアントの LLM を借りる」仕組みで、権限承認・モデル選択・タイムアウトの3点で必ずハマる。この記事ではその実装と回避策をまとめる。

TL;DR

• MCP サーバー内から ctx.session.create_message() を呼ぶと、ホスト側の LLM に推論を依頼できる
• ホストは必ず ユーザー承認(またはポリシー)を挟む ため、サーバー単体ではテストできない
• モデル選択は modelPreferences で「ヒント」を出すだけで確約されない。ホストが無視するケースがある
• タイムアウト・キャンセルはサーバー側で明示的にハンドリングしないと、承認待ちのままプロセスがハングする

手順 / 動かし方

1. サーバー側で Sampling をリクエストする

Python SDK では Context オブジェクト経由で create_message を呼ぶ。

from mcp.server.fastmcp import FastMCP, Context
from mcp.types import SamplingMessage, TextContent

mcp = FastMCP("sampling-demo")

@mcp.tool()
async def summarize_log(log_text: str, ctx: Context) -> str:
    """ログをホストの LLM に要約させるツール"""
    result = await ctx.session.create_message(
        messages=[
            SamplingMessage(
                role="user",
                content=TextContent(
                    type="text",
                    text=f"次のログを3行で要約して:\n\n{log_text}",
                ),
            )
        ],
        max_tokens=300,
        # ヒントであって確約ではない点に注意(後述)
        model_preferences={
            "hints": [{"name": "claude-sonnet"}],
            "intelligencePriority": 0.5,
            "speedPriority": 0.8,
        },
    )
    return result.content.text

ポイントは、ツール自身は推論せず「推論してほしい内容」をホストに投げているだけという点。サーバーは LLM の API キーを一切持たなくていい。

2. ホスト側の承認フローを確認する

Claude Code で動かすと、ツール実行中に Sampling リクエストが来た瞬間、通常のツール承認とは別に「MCP サーバーが LLM 呼び出しを要求しています」という趣旨の確認が入る(実装・バージョンにより挙動は変わる)。ここを自動化テストでスキップしたい場合は、CI では Sampling を使うツールをモックに差し替えるのが現実的。

# テスト用: create_message をモックして承認フローを回避
class FakeSession:
    async def create_message(self, **kwargs):
        return type("R", (), {"content": type("C", (), {"text": "モック要約結果"})()})()

3. 実行結果を確認する

Claude Code から summarize_log ツールを呼び出すと、サーバー内で create_message が実際にホストの LLM を経由して応答を返してくることを、ログ(stderr 経由、後述)で確認できた。

[sampling-demo] create_message request sent (model_hint=claude-sonnet)
[sampling-demo] host approved, response received (287 tokens)

ハマりどころ

① create_message がいつまでも返ってこない(ハング)

エラー文/症状:ツール呼び出し全体がタイムアウトするか、Claude Code 側が固まったように見える。

原因:ホストが承認 UI を出しているのに、非対話環境(headless 実行や CI)だと誰も承認しないため、リクエストが宙に浮く。

回避策:Sampling を使うツールは対話セッション前提で設計し、headless/CI 実行時は該当ツールを呼ばないよう分岐する。もしくは asyncio.wait_for で明示的にタイムアウトを切り、失敗時のフォールバック文言を用意する。

import asyncio

try:
    result = await asyncio.wait_for(
        ctx.session.create_message(messages=[...], max_tokens=300),
        timeout=30,
    )
except asyncio.TimeoutError:
    return "(要約に失敗しました: ホストの承認待ちでタイムアウト)"

② modelPreferences を指定したのに違うモデルが使われる

症状:hints で軽量モデルを指定したつもりが、ホストの既定モデル(ユーザーがセッションで選んでいるモデル)がそのまま使われる。

原因:仕様上 modelPreferences はあくまで「ヒント」であり、ホストが必ず従う保証はない。ホストの実装によっては完全に無視されることもある。

回避策:モデル固定が必須な処理は Sampling に頼らず、サーバー自身が API キーを持って直接 Anthropic API を叩く設計に倒す。Sampling は「ホストのコンテキストと権限をそのまま使い回したい」ケース向けと割り切る。

③ stdout にログを出して JSON-RPC を壊す

エラー文:ホスト側で Unexpected token や MCP 接続断エラーが出る。

原因:MCP の stdio トランスポートは stdout を JSON-RPC 専用に使っているため、print() でデバッグログを出すとメッセージフレームが壊れる。Sampling 導入でリクエスト/レスポンスのやり取りが増え、デバッグしたくて print を仕込んだ瞬間に踏みがちなミス。

回避策:ログは必ず sys.stderr か logging 経由で出す。

import sys

print("[sampling-demo] request sent", file=sys.stderr)

背景・補足

Sampling が用意されている理由は、MCP サーバーが「LLM を呼びたいが、APIキー管理や課金主体はホスト(ユーザー)に持たせたい」というケースに対応するため。サーバー作者が自前で Anthropic API キーを配布・管理する必要がなくなる一方、承認フローやモデル確約がない分、実運用では「重要な処理は Sampling に依存しない」設計判断が必要になる。

まとめ

• Sampling はサーバーがホストの LLM を借りて推論させる仕組みで、API キー管理をホストに委譲できる
• 承認待ちでハングするリスクがあるため、非対話環境ではタイムアウトかツール分岐が必須
• modelPreferences は確約されないので、モデル固定が必要な処理には向かない
• デバッグログは必ず stderr に出す(stdout は JSON-RPC 専用)
• モデル・SDK バージョン依存が大きい機能なので、実装前に手元の MCP ホストでの挙動を必ず確認する