Note
.NETでのセルフホスティング MCP ツールのサポートは近日公開予定です。
Note
現在、Go では、セルフホスティング MCP ツールのサポートは利用できません。
agent-framework-hosting-mcpを使用して、Agent Framework エージェントまたはワークフローをネイティブ モデル コンテキスト プロトコル SDK のツールとして公開します。 パッケージは、Web フレームワークを選択したり、MCP SDK サーバーのライフサイクルをラップしたりすることはありません。アプリケーションは引き続き、 Server、ハンドラー登録、トランスポート、セッション キー ポリシー、認証、承認、デプロイを所有しています。
pip install --pre agent-framework-hosting-mcp
プロトコル境界で変換する
mcp_to_run(...) は、検証済みの MCP ツール引数を Agent Framework メッセージと選択したチャット オプションに変換し、 mcp_from_run(...) 完了した応答をネイティブ MCP ContentBlock 値に変換します。 アプリケーションのツール コントラクトに完全にカスタムのネイティブ スキーマとハンドラーが必要な場合は、次の 2 つの関数を直接使用します。
@server.list_tools()
async def list_tools() -> list[types.Tool]:
"""Return the app-owned native MCP tool definition."""
return [
types.Tool(
name="run_agent_manually",
description=agent.description or "",
inputSchema={
"type": "object",
"properties": {
TASK_ARGUMENT: {
"type": "string",
"description": "The request for the hosted agent.",
},
**CHAT_OPTION_ARGUMENTS,
},
"required": [TASK_ARGUMENT],
"additionalProperties": False,
},
)
]
@server.call_tool()
async def call_tool(name: str, arguments: dict[str, object] | None) -> list[types.ContentBlock]:
"""Convert, run, and render without the agent-backed adapter."""
if name != "run_agent_manually":
raise ValueError(f"Unknown MCP tool: {name}")
run = mcp_to_run(
arguments,
argument_name=TASK_ARGUMENT,
chat_option_arguments=CHAT_OPTION_ARGUMENTS,
)
result = await agent.run(run["messages"], options=run["options"])
return mcp_from_run(result)
chat_option_argumentsにリストされている引数名のみがrun["options"]にコピーされます。他の MCP 引数はメッセージの生表現で使用できますが、モデル クライアントには転送されません。
生成された 1 つのツールとしてエージェントをホストする
AgentMCPTool は、エージェントからネイティブのツール名、説明、スキーマを取得し、一覧表示、解析、実行、結果変換の整合性を保つことで、両者にずれが生じないようにします。
agent_tool = AgentMCPTool(
agent,
name="run_agent",
argument_description="The request for the hosted agent.",
chat_option_parameters={
"reasoning_effort": {
"type": "string",
"enum": ["low", "medium", "high"],
"description": "Optional reasoning effort for models that support it.",
}
},
)
@server.list_tools()
async def list_tools() -> list[types.Tool]:
"""Describe the app-owned MCP tool schema."""
return await agent_tool.list_tools()
@server.call_tool()
async def call_tool(name: str, arguments: dict[str, object] | None) -> list[types.ContentBlock]:
"""Run the app-owned tool with native MCP and Agent Framework values."""
return await agent_tool.call_tool(name, arguments)
AgentMCPTool は、オーバーライドされない限り、エージェントの名前と説明を使用します。
parameters では、未加工の MCP 引数で使用できるアプリ所有の JSON スキーマ プロパティが追加され、 chat_option_parameters は値が Agent Framework チャット オプションに明示的にコピーされるプロパティを追加します。
呼び出しごとにセッションを保持する
既存の AgentState と session_id_parameter を渡すと、同じ不透明な、アプリで定義された session_id を使った繰り返し呼び出しで、1 つの会話を継続できるようになります:
session_locks: dict[str, asyncio.Lock] = {}
@server.list_tools()
async def list_tools() -> list[types.Tool]:
"""Return the agent-derived MCP tool definition."""
return await agent_tool.list_tools()
@server.call_tool()
async def call_tool(name: str, arguments: dict[str, object] | None) -> list[types.ContentBlock]:
"""Serialize calls per app-owned session before using ``AgentState``."""
session_id = arguments.get("session_id") if arguments else None
if not isinstance(session_id, str) or not session_id:
raise ValueError("MCP tool argument 'session_id' must be a non-empty string.")
lock = session_locks.setdefault(session_id, asyncio.Lock())
async with lock:
return await agent_tool.call_tool(name, arguments)
AgentMCPTool は、 AgentState セッションの取得/実行/設定シーケンスのみを実行します。アプリケーションはセッション識別子を認証または承認し、同じセッションの同時呼び出しをシリアル化する必要があります。これは、サンプルがセッションごとの asyncio.Lockで行う場合と同じです。 これは previous_response_idスタイルの分岐ではありません。会話をフォークする必要があるアプリケーションでは、個別のソース ID と宛先 ID を受け入れ、ソース セッションをコピーし、結果を宛先キーの下に格納する必要があります。
ワークフローをツールとしてホストする
WorkflowMCPTool は、ワークフローの start-executor 入力型から 1 つのネイティブ MCP ツールを派生させ、完成したワークフロー出力を変換します。 データクラス、Pydantic、およびその他のオブジェクト型の入力は、最上位の MCP 引数になります。プリミティブ入力は、構成可能な引数名でラップされます。
server = Server("agent-framework-hosting-mcp-workflow-sample")
workflow_tool = WorkflowMCPTool(
WorkflowState(create_workflow, cache_target=False),
name="draft_content",
)
ワークフロー インスタンスは実行状態を維持するため、独立した呼び出しを必要とするアプリケーションは、上記のようにWorkflowStateを備えたcache_target=False ファクトリを提供する必要があります。 チェックポイントの復元、ループ内の人間の応答、継続識別子は、アプリケーション所有のままです。ワークフローが外部入力を要求した場合、アダプターは空の成功したツールの結果を返す代わりに発生します。
装飾された関数からスキーマを派生させる FastMCP バリアントを含む、実行可能なサーバーの完全なセットについては、 MCP ホスティングのサンプルを参照してください。
Important
MCP セッション識別子とアプリ定義の session_id 引数を信頼されていない入力として扱います。 セッション状態の読み込みまたは保存を行う前に呼び出し元を認証および承認し、未加工の値ではなく、認証済みのテナント、ユーザー、またはワークスペースから永続的なパーティション分割を派生させます。
次のステップ
より深く進む: