Managed Agentsセルフホスト型サンドボックス

セルフホスト型サンドボックスでのカスタムツール

Copy page



セルフホスト型サンドボックスのワーカーからカスタムツールを提供し、トンネルを実行せずにネットワーク内のMCPサーバーをカスタムツールとしてラップします。

カスタムツールは、独自のコードが実行するツールです。エージェントは agent.custom_tool_use イベントを発行し、対応する user.custom_tool_result を待ちます。ワーカーがそのコードになることができます。ワーカーはサンドボックス内で実行されるため、ツールはサンドボックスに設定した内部サービス、認証情報、ネットワークの「egress」（エグレス）にアクセスでき、それ以外にはアクセスしません。

環境キーによってカスタムツールの結果の送信が認可されるため、Claude APIキーをワーカーホストに置く必要はありません。

カスタムツールを提供する

1. 1

   エージェントでツールを宣言する

   エージェントの tools に custom エントリを追加し、その name をワーカーが登録するツールと一致させます。宣言の完全な形式については、カスタムツールを参照してください。

   {
     "type": "custom",
     "name": "get_order_status",
     "description": "Look up an order in the internal fulfillment system by order ID.",
     "input_schema": {
       "type": "object",
       "properties": {
         "order_id": { "type": "string", "description": "The order ID" }
       },
       "required": ["order_id"]
     }
   }

   
2. 2

   ワーカーに実装を登録する

   組み込みツールセットと並べて、ワーカーの tools ファクトリ（EnvironmentWorkerを参照）を通じてツールを渡します。

   PythonTypeScriptC#GoJavaPHPRuby

   

   import asyncio
   import os
   from anthropic import AsyncAnthropic, beta_async_tool
   from anthropic.lib.environments import EnvironmentWorker
   from anthropic.lib.tools.agent_toolset import beta_agent_toolset_20260401
   
   
   @beta_async_tool
   async def get_order_status(order_id: str) -> str:
       """Look up an order in the internal fulfillment system by order ID."""
       # ワーカーホスト上で実行されます:サンドボックスから到達できるものは何でも呼び出せます。
       return f"Order {order_id}: shipped"
   
   
   async def main() -> None:
       environment_key = os.environ["ANTHROPIC_ENVIRONMENT_KEY"]
       environment_id = os.environ["ANTHROPIC_ENVIRONMENT_ID"]
       async with AsyncAnthropic(auth_token=environment_key) as client:
           await EnvironmentWorker(
               client,
               environment_id=environment_id,
               environment_key=environment_key,
               workdir="/workspace",
               tools=lambda env: [*beta_agent_toolset_20260401(env), get_order_status],
           ).run()
   
   
   asyncio.run(main())

ワーカーは、自身に登録されたツールにのみ応答します。ツールがエージェントで宣言されているものの、それを提供するワーカーやクライアントが存在しない場合、セッションは requires_action の停止理由で一時停止します。何かが結果を送信するまで、セッションは一時停止したままになります。イベントフローについては、カスタムツール呼び出しの処理を参照してください。

MCPサーバーをカスタムツールとしてラップする

MCPコネクタは、Anthropic側からMCPサーバーに接続します。そのため、サーバーは、直接またはMCPトンネルを介して、Anthropicが到達できるHTTPエンドポイントを公開する必要があります。

自社のネットワークからしか到達できないサーバーを使用するには、代わりにワーカーをMCPクライアントにし、サーバーのツールをカスタムツールとして宣言します。MCPサーバーには、ネットワーク外部からのインバウンド接続は必要ありません。Anthropicが受け取るのは、エージェントで宣言したツール定義、各呼び出しの入力、そしてワーカーが送り返す結果です。

実行時、モデルはラップされたツールを他のカスタムツールと同様に呼び出します。

1. エージェントが agent.custom_tool_use イベントを発行します。
2. サンドボックス内のワーカーが、開いているMCPセッションを介して、ネットワーク上のサーバーに呼び出しを転送します。
3. ワーカーがサーバーの応答を user.custom_tool_result として送信します。

MCP SDKをインストールする

SDKのクライアントサイドのMCPヘルパーは、サーバーのツールをワーカーが受け付ける実行可能なツールに変換します。Anthropic SDKと併せてMCP SDKをインストールしてください：pip install "anthropic[mcp]" "mcp>=1.24"。

例では認証なしで接続しています。認証情報を送信するには、MCPトランスポートに渡す http_client を設定してください。

ツールを宣言して提供する

1. 1

   エージェントでサーバーのツールを宣言する

   MCPサーバーのツールを一覧表示し、それぞれを custom ツールとして宣言します。MCPの name、description、inputSchema は、カスタムツールのフィールドに1対1で対応します。サーバーがツール一覧をページ分割している場合は、すべてのページを宣言してください。ワーカーも同じページを一覧表示する必要があります。

   PythonTypeScriptC#GoJavaPHPRuby

   

   import asyncio
   from typing import Any, cast
   from anthropic import AsyncAnthropic
   from anthropic.types.beta import BetaManagedAgentsCustomToolParams
   from mcp import ClientSession, types
   # mcp >= 1.24 が必要です（streamablehttp_client が streamable_http_client に改名されました）。
   from mcp.client.streamable_http import streamable_http_client
   
   MCP_SERVER_URL = "http://mcp.internal.example.com:8000/mcp"
   
   
   def to_custom_tool(tool: types.Tool) -> BetaManagedAgentsCustomToolParams:
       # MCPのフィールドはカスタムツール宣言に1対1で対応します。cast は
       # スキーマ辞書をそのままSDKの型付きパラメータに渡します。
       return {
           "type": "custom",
           "name": tool.name,
           "description": tool.description or tool.name,
           "input_schema": cast(Any, tool.inputSchema),
       }
   
   
   async def main() -> None:
       # これはワーカーホストではなく、エージェントを作成する場所で実行してください。
       # Claude APIキー（ANTHROPIC_API_KEY）で認証します。
       async with (
           streamable_http_client(MCP_SERVER_URL) as (read, write, _),
           ClientSession(read, write) as mcp_session,
           AsyncAnthropic() as client,
       ):
           await mcp_session.initialize()
           listed = await mcp_session.list_tools()
           agent = await client.beta.agents.create(
               name="Internal tools agent",
               model="claude-opus-5-5",
               tools=[
                   {"type": "agent_toolset_20260401"},
                   *[to_custom_tool(tool) for tool in listed.tools],
               ],
           )
           print(agent.id)
   
   
   asyncio.run(main())

2. 2

   ワーカーからツールを提供する

   起動時に同じMCPサーバーに接続し、async_mcp_tool でそのツールを変換して、beta_agent_toolset_20260401 と並べて登録します。ワーカーが稼働している間は、1つのMCPセッションを開いたままにしてください。

   PythonTypeScriptC#GoJavaPHPRuby

   

   import asyncio
   import os
   from datetime import timedelta
   from anthropic import AsyncAnthropic
   from anthropic.lib.environments import EnvironmentWorker
   from anthropic.lib.tools.agent_toolset import beta_agent_toolset_20260401
   from anthropic.lib.tools.mcp import async_mcp_tool
   from mcp import ClientSession
   # mcp >= 1.24 が必要です(streamablehttp_client が streamable_http_client に改名されました)。
   from mcp.client.streamable_http import streamable_http_client
   
   MCP_SERVER_URL = "http://mcp.internal.example.com:8000/mcp"
   
   
   async def main() -> None:
       environment_key = os.environ["ANTHROPIC_ENVIRONMENT_KEY"]
       environment_id = os.environ["ANTHROPIC_ENVIRONMENT_ID"]
       # 起動時に一度だけ MCP サーバーに接続し、ワーカーの存続期間中は
       # セッションを開いたままにします。タイムアウトにより、ハングしたツール呼び出しは
       # 停止したままになるのではなくエラー結果として返されます。
       async with (
           streamable_http_client(MCP_SERVER_URL) as (read, write, _),
           ClientSession(read, write, read_timeout_seconds=timedelta(seconds=60)) as mcp_session,
           AsyncAnthropic(auth_token=environment_key) as client,
       ):
           await mcp_session.initialize()
           listed = await mcp_session.list_tools()
           mcp_tools = [async_mcp_tool(tool, mcp_session) for tool in listed.tools]
           await EnvironmentWorker(
               client,
               environment_id=environment_id,
               environment_key=environment_key,
               workdir="/workspace",
               tools=lambda env: [*beta_agent_toolset_20260401(env), *mcp_tools],
           ).run()
   
   
   asyncio.run(main())

制限と動作

ツールは宣言されるものであり、実行時に検出されるものではない

ワーカーは起動時に一度だけMCPサーバーのツールを一覧表示し、実行中のセッションにツールを追加することはできません。サーバーのツールが変更された場合は、次の手順を実行します。

1. エージェントで、またはエージェント設定の更新を通じてアイドル状態のセッションで、ツールを再度宣言します。
2. ワーカーを再起動します。

宣言はManaged Agents APIに適合している必要がある

MCPヘルパーはサーバーの名前と説明を保持し、ほとんどのスキーマは変更されずにそのまま渡されます。宣言が次のいずれかのルールに違反する場合は、名前の変更、切り詰め、またはインライン化を行ってください。

次の2つのケースでは追加の作業が必要です。

• 2つのサーバーが同じツール名を公開している場合： プレフィックス付きの名前でラッパーを自分で定義し、そのラッパーからサーバーの元のツール名を呼び出すようにします。
• pydanticなどのジェネレーターがスキーマを $defs に分解している場合： ツールを宣言する前に、それらのスキーマをインライン化します。

ツールの失敗はエラーのツール結果として表面化する

MCPサーバーがツールエラーを報告すると、ワーカーはモデルが対応できるエラーのツール結果を送信します。オーディオブロックやリソースリンクなど、ツール結果に相当するものがないMCPコンテンツも、エラーとして表面化します。

Pythonワーカーの例で read_timeout_seconds を使用しているように、MCPクライアントにタイムアウトを設定すると、より迅速かつ明確に失敗させることができます。タイムアウトがない場合に何が起こるかについては、ラップされたMCPツール呼び出しがハングするを参照してください。

自社で運用しているか信頼できるサーバーのみをラップする

ラップされたツールの名前、説明、結果は、他のツールと同様にモデルのコンテキストに入ります。これらは信頼できない入力であり、ワーカーホスト上の bash を含め、エージェントが他のツールで行う動作に影響を与える可能性があります。エージェントに使用させるつもりのツールのみを宣言してください。

権限ポリシーは適用されない

権限ポリシーは、組み込みツールセットとMCPツールセットを管理します。ワーカーはモデルが行うラップされたツール呼び出しをすべて実行するため、承認ステップが必要な場合は独自のツールコードに組み込んでください。