AgentCore ランタイムに MCP サーバーをデプロイする
Amazon Bedrock AgentCore ランタイムでは、モデルコンテキストプロトコル (MCP) サーバーを AgentCore ランタイムにデプロイして実行できます。このガイドでは、最初の MCP サーバーの作成、テスト、デプロイについて説明します。
例については、https://github.com/awslabs/amazon-bedrock-agentcore-samples/tree/main/01-tutorials/01-AgentCore-runtime/02-hosting-MCP-server
このセクションでは、以下を行います。
-
ツールを使用して MCP サーバーを作成する方法
-
サーバーをローカルでテストする方法
-
サーバーを にデプロイする方法 AWS
-
デプロイされたサーバーを呼び出す方法
MCP の詳細については、「MCP プロトコル契約」を参照してください。
トピック
Amazon Bedrock AgentCore が MCP をサポートする方法
MCP プロトコルを使用して Amazon Bedrock AgentCore ランタイムを設定する場合、このサービスは、最も公式な MCP サーバー SDK 0.0.0.0:8000/mcp でサポートされているデフォルトパスであるパス で MCP サーバーコンテナが利用可能であることを期待します。 SDKs
Amazon Bedrock AgentCore は、ステートレスとステートフルの両方のストリーミング可能な HTTP MCP サーバーをサポートしています。デフォルトでは、基本的な MCP サーバーにはステートレスモード () stateless_http=True が推奨されます。プラットフォームは、リクエストなしで自動的に Mcp-Session-Idヘッダーを追加するため、MCP クライアントは同じ Amazon Bedrock AgentCore ランタイムセッションへの接続継続性を維持できます。
マルチターンインタラクション (誘発）、LLM 生成コンテンツ (サンプリング）、または進行状況通知を必要とする MCP サーバーの場合、ステートフルモード () stateless_http=False によりこれらの機能が有効になります。ステートフルモードでは、ランタイムは同じ呼び出し内のリクエスト間で MCP セッション状態を保持します。詳細については、「ステートフル MCP サーバー機能」を参照してください。
InvokeAgentRuntime API のペイロードは直接渡されるため、MCP などのプロトコルの RPC メッセージを簡単にプロキシできます。
前提条件
-
Python 3.10 以上がインストールされ、Python の基本的な理解
-
適切なアクセス許可とローカル認証情報が設定されている AWS アカウント
ステップ 1: MCP サーバーを作成する
必要なパッケージをインストールする
まず、MCP パッケージをインストールします。
pip install mcp
最初の MCP サーバーを作成する
という名前の新しいファイルを作成しますmy_mcp_server.py。
# my_mcp_server.py from mcp.server.fastmcp import FastMCP from starlette.responses import JSONResponse mcp = FastMCP(host="0.0.0.0", stateless_http=True) @mcp.tool() def add_numbers(a: int, b: int) -> int: """Add two numbers together""" return a + b @mcp.tool() def multiply_numbers(a: int, b: int) -> int: """Multiply two numbers together""" return a * b @mcp.tool() def greet_user(name: str) -> str: """Greet a user by name""" return f"Hello, {name}! Nice to meet you." if __name__ == "__main__": mcp.run(transport="streamable-http")
コードについて
-
FastMCP : ツールをホストできる MCP サーバーを作成します
-
@mcp.tool() : Python 関数を MCP ツールに変換するデコレータ
-
ツール: さまざまなタイプのオペレーションを示す 3 つのシンプルなツール
-
stateless_http=True : 基本的な MCP サーバーのデフォルトであるステートレスモードでサーバーを設定します。
ヒント
マルチターンインタラクション (誘発) または LLM 生成コンテンツ (サンプリング) を必要とする MCP サーバーの場合は、 stateless_http=False を使用してステートフルモードを有効にします。ステートフル MCP サーバーは、同じツール呼び出し内の複数のリクエストにわたってセッションコンテキストを維持します。詳細については、「ステートフル MCP サーバー機能」を参照してください。
ステップ 2: MCP サーバーをローカルでテストする
MCP サーバーを起動する
MCP サーバーをローカルで実行します。
python my_mcp_server.py
サーバーがポート で実行されていることを示す出力が表示されます8000。
MCP クライアントでテストする
新しいターミナルから新しいファイルを作成しmy_mcp_client.py、 を使用して実行します。 python my_mcp_client.py
# my_mcp_client.py import asyncio from mcp import ClientSession from mcp.client.streamable_http import streamablehttp_client async def main(): mcp_url = "http://localhost:8000/mcp" headers = {} async with streamablehttp_client(mcp_url, headers, timeout=120, terminate_on_close=False) as ( read_stream, write_stream, _, ): async with ClientSession(read_stream, write_stream) as session: await session.initialize() tool_result = await session.list_tools() print(tool_result) asyncio.run(main())
MCP Inspector を使用したローカルテストで説明されているように、MCP Inspector を使用してサーバーをテストすることもできます。
ステップ 3: MCP サーバーを にデプロイする AWS
デプロイツールをインストールする
AgentCore CLI をインストールします。
npm install -g @aws/agentcore
AgentCore CLI を使用して、エージェントを AgentCore ランタイムにデプロイします。
次の構造でプロジェクトフォルダを作成します。
## Project Folder Structure your_project_directory/ ├── mcp_server.py # Your main agent code ├── requirements.txt # Dependencies for your agent └── __init__.py # Makes the directory a Python package
という名前の新しいファイルを作成しrequirements.txt、そのファイルに以下を追加します。
mcp
requirements.txt は、エージェントが AgentCore ランタイムにデプロイするために必要な要件を指定します。
デプロイ用のプロジェクトを作成する
プロジェクトを作成する前に、「認証用の Cognito ユーザープールを設定する」の説明に従って、認証用の Cognito ユーザープールを設定する必要があります。これにより、デプロイされたサーバーへの安全なアクセスに必要な OAuth トークンが提供されます。
注記
2025 年 10 月 7 日以降、Amazon Bedrock AgentCore は OAuth 認証を使用するときにワークロード ID のアクセス許可にサービスにリンクされたロールを使用します。この変更の詳細については、「 Identity service-linked role」を参照してください。
認証を設定したら、MCP プロトコルを使用して新しいプロジェクトをスキャフォールドします。
agentcore create --protocol MCP
インタラクティブプロンプトに従って、プロジェクト名を指定します。CLI は、agentcore/agentcore.json設定ファイルを含むプロジェクト構造を足場します。生成されたプロジェクトのエージェントコードディレクトリにmy_mcp_server.pyファイルをコピーし、 のエントリポイントがサーバーファイルagentcore/agentcore.jsonを指すことを確認します。
にデプロイする AWS
エージェントをデプロイします。
agentcore deploy
このコマンドでは、次の操作を行います。
-
エージェントコードと依存関係をパッケージ化する
-
デプロイアーティファクトを Amazon S3 にアップロードする
-
Amazon Bedrock AgentCore ランタイムを作成する
-
エージェントを にデプロイする AWS
デプロイ後、エージェントランタイム ARN は次のようになります。
arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_mcp_server-xyz123
ステップ 4: デプロイされた MCP サーバーを呼び出す
MCP クライアントでテストする (リモート)
テストする前に、次の環境変数を設定します。
-
エージェント ARN を環境変数としてエクスポートします。
export AGENT_ARN="agent_arn" -
ベアラートークンを環境変数としてエクスポートします。
export BEARER_TOKEN="bearer_token"
Accept ヘッダーを渡す場合は、MCPapplication/jsonと ですtext/event-stream。
を使用して新しいファイルを作成しmy_mcp_client_remote.py、実行する python my_mcp_client_remote.py
import asyncio import os import sys from mcp import ClientSession from mcp.client.streamable_http import streamablehttp_client async def main(): agent_arn = os.getenv('AGENT_ARN') bearer_token = os.getenv('BEARER_TOKEN') if not agent_arn or not bearer_token: print("Error: AGENT_ARN or BEARER_TOKEN environment variable is not set") sys.exit(1) encoded_arn = agent_arn.replace(':', '%3A').replace('/', '%2F') mcp_url = f"https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/{encoded_arn}/invocations?qualifier=DEFAULT" headers = {"authorization": f"Bearer {bearer_token}","Content-Type":"application/json"} print(f"Invoking: {mcp_url}, \nwith headers: {headers}\n") async with streamablehttp_client(mcp_url, headers, timeout=120, terminate_on_close=False) as ( read_stream, write_stream, _, ): async with ClientSession(read_stream, write_stream) as session: await session.initialize() tool_result = await session.list_tools() print(tool_result) asyncio.run(main())
MCP Inspector を使用してデプロイされたサーバーをテストすることもできます。詳細については、「MCP Inspector を使用したリモートテスト」を参照してください。
OAuth 設定エージェントの認証エラーレスポンス
OAuth 設定のエージェントは、RFC 6749 (OAuth 2.0)
401 未承認 - 認証がありません
認可ヘッダーにベアラートークンが指定されていない場合、レスポンスは次のようになります。
HTTP/1.1 401 Unauthorized WWW-Authenticate: Bearer resource_metadata="https://bedrock-agentcore.{region}.amazonaws.com/runtimes/{ESCAPED_ARN}/invocations/.well-known/oauth-protected-resource?qualifier={QUALIFIER}"
Auth0 を使用したエンドツーエンドのフロー
このセクションでは、ID プロバイダーとして Auth0 を使用した OAuth 認証を示します。この例では、動的クライアント登録 (DCR) をサポートしているため、Auth0 を使用します。これにより、クライアントが実行時にプログラムで登録できるため、クライアントのセットアッププロセスが簡素化されます。
ステップ 1 - ステップ 3: MCP サーバーを作成してテストする
ステップ 1: MCP サーバーを作成するステップ 3: MCP サーバーを にデプロイして MCP サーバーを作成してテストするステップ 1～3 AWSに従います。
ステップ 4: Auth0 アプリケーションを作成する
Okta による Auth0 の設定手順に従います。 Auth0
動的クライアント登録を有効にする:
-
ダッシュボード → 設定 → 詳細
-
「OIDC 動的アプリケーション登録」の切り替え → ON
-
変更を保存する
詳細については、Auth0 Dynamic Client Registration documentation
ステップ 5: デプロイするプロジェクトを作成する
認証を設定したら、MCP プロトコルを使用して新しいプロジェクトをスキャフォールドします。
agentcore create --protocol MCP
インタラクティブプロンプトに従って、プロジェクト名を指定します。CLI は、agentcore/agentcore.json設定ファイルを含むプロジェクト構造を足場します。生成されたプロジェクトのエージェントコードディレクトリにmy_mcp_server.pyファイルをコピーし、 のエントリポイントがサーバーファイルagentcore/agentcore.jsonを指すことを確認します。
ステップ 6: にデプロイする AWS
エージェントをデプロイします。
agentcore deploy
このコマンドでは、次の操作を行います。
-
エージェントコードと依存関係をパッケージ化する
-
デプロイアーティファクトを Amazon S3 にアップロードする
-
Amazon Bedrock AgentCore ランタイムを作成する
-
エージェントを にデプロイする AWS
デプロイ後、エージェントランタイム ARN は次のようになります。
arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_mcp_server-xyz123
ステップ 7: デプロイされた エージェントを呼び出す
このクライアントは、Auth0-specific0 固有の変更を加えた公式 MCP SDK simple-auth-client の例
注記
動的クライアント登録で Auth0 を使用する場合は、JWT トークンを受信するための認可リクエストに audienceパラメータを含める必要があります。このパラメータがない場合、Auth0 は標準 JWT トークンの代わりに不透明トークンまたは JWE (暗号化) トークンを返します。MCP SDK は OAuth 2.0 の resourceパラメータ (RFC 8707) を送信しますが、Auth0 には JWT トークンの OIDC audienceパラメータが必要です。どちらのパラメータも同様の目的を果たしますが、Auth0 は audience を優先します。詳細については、Auth0 Community - JWT tokens with Dynamic Application Registration
mcp_auth0_client.py という名前のファイルを作成し、次のコードを記述します。このクライアントは、対象者パラメータを含む Auth0-specific要件を処理します。
注記
このコードには、すべての HTTP リクエストに User-Agent ヘッダーを挿入するための httpx パッチ適用が含まれています。MCP Python SDK では現在、HTTP リクエストに User-Agent ヘッダーが含まれていないため、User-Agent ヘッダーを必要とする AWS WAF ルールに問題が発生する可能性があります。詳細については、「MCP Python SDK 問題 #1664
#!/usr/bin/env python3 """ MCP client with OAuth authentication support for Auth0. Based on the official MCP SDK simple-auth-client example with Auth0 compatibility. Adds support for Auth0's 'audience' parameter requirement. Usage: # Required export AGENT_ARN="arn:aws:bedrock:us-west-2:123456789012:agent/ABCD1234" # Required for Auth0 export AUTH0_API_IDENTIFIER="your-api-identifier" # Optional - custom endpoint for beta/dev environments export CUSTOM_ENDPOINT="https://beta.example.com" python mcp_auth0_client.py The client will automatically: - Encode the Agent ARN for use in the URL - Construct the MCP invocation endpoint URL - Add Auth0 'audience' parameter to authorization requests (when using Auth0) - Work with any OAuth 2.0 compliant identity provider """ import asyncio import httpx import os import threading import time import webbrowser from datetime import timedelta from http.server import BaseHTTPRequestHandler, HTTPServer from typing import Any from urllib.parse import parse_qs, urlencode, urlparse, urlunparse # Patch httpx at the request level to inject User-Agent header # This ensures ALL HTTP requests have the User-Agent header, including OAuth discovery calls _original_httpx_request = httpx.Request.__init__ def _patched_httpx_request_init(self, method, url, *args, **kwargs): """Patched Request.__init__ that injects User-Agent header into all HTTP requests.""" # Get or create headers headers = kwargs.get('headers') if headers is None: headers = {} kwargs['headers'] = headers # Convert to mutable dict if needed if not isinstance(headers, dict): headers = dict(headers) kwargs['headers'] = headers # Inject User-Agent if not present (case-insensitive check) if 'User-Agent' not in headers and 'user-agent' not in headers: headers['User-Agent'] = 'python-mcp-sdk/1.0 (BedrockAgentCore-Runtime)' # Call original __init__ _original_httpx_request(self, method, url, *args, **kwargs) # Apply the patch globally before importing MCP modules httpx.Request.__init__ = _patched_httpx_request_init # Now import MCP modules - they will use patched httpx from mcp.client.auth import OAuthClientProvider, TokenStorage from mcp.client.session import ClientSession from mcp.client.sse import sse_client from mcp.client.streamable_http import streamablehttp_client from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken class InMemoryTokenStorage(TokenStorage): """Simple in-memory token storage implementation.""" def __init__(self): self._tokens: OAuthToken | None = None self._client_info: OAuthClientInformationFull | None = None async def get_tokens(self) -> OAuthToken | None: return self._tokens async def set_tokens(self, tokens: OAuthToken) -> None: self._tokens = tokens async def get_client_info(self) -> OAuthClientInformationFull | None: return self._client_info async def set_client_info(self, client_info: OAuthClientInformationFull) -> None: self._client_info = client_info class CallbackHandler(BaseHTTPRequestHandler): """Simple HTTP handler to capture OAuth callback.""" def __init__(self, request, client_address, server, callback_data): """Initialize with callback data storage.""" self.callback_data = callback_data super().__init__(request, client_address, server) def do_GET(self): """Handle GET request from OAuth redirect.""" parsed = urlparse(self.path) query_params = parse_qs(parsed.query) if "code" in query_params: self.callback_data["authorization_code"] = query_params["code"][0] self.callback_data["state"] = query_params.get("state", [None])[0] self.send_response(200) self.send_header("Content-type", "text/html") self.end_headers() self.wfile.write(b""" <html> <body> <h1>Authorization Successful!</h1> <p>You can close this window and return to the terminal.</p> <script>setTimeout(() => window.close(), 2000);</script> </body> </html> """) elif "error" in query_params: self.callback_data["error"] = query_params["error"][0] self.send_response(400) self.send_header("Content-type", "text/html") self.end_headers() self.wfile.write( f""" <html> <body> <h1>Authorization Failed</h1> <p>Error: {query_params["error"][0]}</p> <p>You can close this window and return to the terminal.</p> </body> </html> """.encode() ) else: self.send_response(404) self.end_headers() def log_message(self, format, *args): """Suppress default logging.""" pass class CallbackServer: """Simple server to handle OAuth callbacks.""" def __init__(self, port=3030): self.port = port self.server = None self.thread = None self.callback_data = {"authorization_code": None, "state": None, "error": None} def _create_handler_with_data(self): """Create a handler class with access to callback data.""" callback_data = self.callback_data class DataCallbackHandler(CallbackHandler): def __init__(self, request, client_address, server): super().__init__(request, client_address, server, callback_data) return DataCallbackHandler def start(self): """Start the callback server in a background thread.""" handler_class = self._create_handler_with_data() self.server = HTTPServer(("localhost", self.port), handler_class) self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) self.thread.start() print(f"🖥️ Started callback server on http://localhost:{self.port}") def stop(self): """Stop the callback server.""" if self.server: self.server.shutdown() self.server.server_close() if self.thread: self.thread.join(timeout=1) def wait_for_callback(self, timeout=300): """Wait for OAuth callback with timeout.""" start_time = time.time() while time.time() - start_time < timeout: if self.callback_data["authorization_code"]: return self.callback_data["authorization_code"] elif self.callback_data["error"]: raise Exception(f"OAuth error: {self.callback_data['error']}") time.sleep(0.1) raise Exception("Timeout waiting for OAuth callback") def get_state(self): """Get the received state parameter.""" return self.callback_data["state"] def add_auth0_audience_parameter(authorization_url: str, audience: str) -> str: """ Add Auth0 'audience' parameter to authorization URL. Auth0 requires the 'audience' parameter to identify which API's token settings to use. Without it, Auth0 returns opaque tokens or JWE instead of JWT. This function properly adds the audience parameter while preserving all existing query parameters (including the OAuth 'resource' parameter). Args: authorization_url: The authorization URL from the OAuth flow audience: The Auth0 API identifier (e.g., "runtime-api") Returns: Modified URL with audience parameter added Reference: https://auth0.com/docs/secure/tokens/access-tokens/get-access-tokens """ # Only apply to Auth0 URLs that don't already have audience if 'auth0.com' not in authorization_url or 'audience=' in authorization_url: return authorization_url # Parse URL and query parameters parsed = urlparse(authorization_url) query_params = parse_qs(parsed.query, keep_blank_values=True) # Add audience parameter query_params['audience'] = [audience] # Rebuild URL with new parameter new_query = urlencode(query_params, doseq=True) return urlunparse(( parsed.scheme, parsed.netloc, parsed.path, parsed.params, new_query, parsed.fragment )) class SimpleAuthClient: """Simple MCP client with Auth0 OAuth support.""" def __init__( self, server_url: str, transport_type: str = "streamable-http", auth0_audience: str | None = None, ): self.server_url = server_url self.transport_type = transport_type self.auth0_audience = auth0_audience self.session: ClientSession | None = None async def connect(self): """Connect to the MCP server.""" print(f"🔗 Attempting to connect to {self.server_url}...") try: callback_server = CallbackServer(port=3030) callback_server.start() async def callback_handler() -> tuple[str, str | None]: """Wait for OAuth callback and return auth code and state.""" print("⏳ Waiting for authorization callback...") try: auth_code = callback_server.wait_for_callback(timeout=300) return auth_code, callback_server.get_state() finally: callback_server.stop() client_metadata_dict = { "client_name": "MCP Auth0 Client", "redirect_uris": ["http://localhost:3030/callback"], "grant_types": ["authorization_code", "refresh_token"], "response_types": ["code"], } async def redirect_handler(authorization_url: str) -> None: """Redirect handler that opens the URL in a browser with Auth0 audience parameter.""" # Add Auth0 audience parameter if configured if self.auth0_audience: authorization_url = add_auth0_audience_parameter( authorization_url, self.auth0_audience ) webbrowser.open(authorization_url) print("\n🔧 Creating OAuth client provider...") # Create OAuth authentication handler # Note: httpx.AsyncClient is globally patched to inject User-Agent header oauth_auth = OAuthClientProvider( server_url=self.server_url, client_metadata=OAuthClientMetadata.model_validate(client_metadata_dict), storage=InMemoryTokenStorage(), redirect_handler=redirect_handler, callback_handler=callback_handler, ) print("🔧 OAuth client provider created successfully") # Create transport with auth handler based on transport type if self.transport_type == "sse": print("📡 Opening SSE transport connection with auth...") async with sse_client( url=self.server_url, auth=oauth_auth, timeout=60, ) as (read_stream, write_stream): await self._run_session(read_stream, write_stream, None) else: print("📡 Opening StreamableHTTP transport connection with auth...") async with streamablehttp_client( url=self.server_url, auth=oauth_auth, timeout=timedelta(seconds=60), ) as (read_stream, write_stream, get_session_id): await self._run_session(read_stream, write_stream, get_session_id) except Exception as e: print(f"❌ Failed to connect: {e}") import traceback traceback.print_exc() async def _run_session(self, read_stream, write_stream, get_session_id): """Run the MCP session with the given streams.""" print("🤝 Initializing MCP session...") async with ClientSession(read_stream, write_stream) as session: self.session = session print("⚡ Starting session initialization...") await session.initialize() print("✨ Session initialization complete!") print(f"\n✅ Connected to MCP server at {self.server_url}") if get_session_id: session_id = get_session_id() if session_id: print(f"Session ID: {session_id}") # Run interactive loop await self.interactive_loop() async def list_tools(self): """List available tools from the server.""" if not self.session: print("❌ Not connected to server") return try: result = await self.session.list_tools() if hasattr(result, "tools") and result.tools: print("\n📋 Available tools:") for i, tool in enumerate(result.tools, 1): print(f"{i}. {tool.name}") if tool.description: print(f" Description: {tool.description}") print() else: print("No tools available") except Exception as e: print(f"❌ Failed to list tools: {e}") async def call_tool(self, tool_name: str, arguments: dict[str, Any] | None = None): """Call a specific tool.""" if not self.session: print("❌ Not connected to server") return try: result = await self.session.call_tool(tool_name, arguments or {}) print(f"\n🔧 Tool '{tool_name}' result:") if hasattr(result, "content"): for content in result.content: if content.type == "text": print(content.text) else: print(content) else: print(result) except Exception as e: print(f"❌ Failed to call tool '{tool_name}': {e}") async def interactive_loop(self): """Run interactive command loop.""" print("\n🎯 Interactive MCP Client") print("Commands:") print(" list - List available tools") print(" call <tool_name> [args] - Call a tool") print(" quit - Exit the client") print() while True: try: command = input("mcp> ").strip() if not command: continue if command == "quit": break elif command == "list": await self.list_tools() elif command.startswith("call "): parts = command.split(maxsplit=2) tool_name = parts[1] if len(parts) > 1 else "" if not tool_name: print("❌ Please specify a tool name") continue # Parse arguments (simple JSON-like format) arguments = {} if len(parts) > 2: import json try: arguments = json.loads(parts[2]) except json.JSONDecodeError: print("❌ Invalid arguments format (expected JSON)") continue await self.call_tool(tool_name, arguments) else: print("❌ Unknown command. Try 'list', 'call <tool_name>', or 'quit'") except KeyboardInterrupt: print("\n\n👋 Goodbye!") break except EOFError: break async def main(): """Main entry point.""" # Get Agent ARN from environment agent_arn = os.getenv("AGENT_ARN") if not agent_arn: print("❌ Please set AGENT_ARN environment variable") print("Example: export AGENT_ARN='arn:aws:bedrock:us-west-2:123456789012:agent/ABCD1234'") return # Encode the ARN for use in URL encoded_arn = agent_arn.replace(':', '%3A').replace('/', '%2F') # Get base URL - use custom endpoint or default to production base_endpoint = os.getenv("CUSTOM_ENDPOINT", "https://bedrock-agentcore.us-west-2.amazonaws.com") # Construct MCP URL from encoded ARN (no qualifier - SDK discovers it from PRM API) server_url = f"{base_endpoint}/runtimes/{encoded_arn}/invocations" # Get Auth0 configuration (required only for Auth0) auth0_audience = os.getenv("AUTH0_API_IDENTIFIER") # Get optional transport type transport_type = os.getenv("MCP_TRANSPORT_TYPE", "streamable-http") print("🚀 MCP Auth0 Client") print(f"Agent ARN: {agent_arn}") print(f"Endpoint: {base_endpoint}") print(f"Connecting to: {server_url}") print(f"Transport type: {transport_type}") if auth0_audience: print(f"Auth0 audience: {auth0_audience}") # Start connection flow - OAuth will be handled automatically client = SimpleAuthClient( server_url, transport_type, auth0_audience, ) await client.connect() def cli(): """CLI entry point for uv script.""" asyncio.run(main()) if __name__ == "__main__": cli()
クライアントを使用するには:
-
必要な環境変数を設定します。
export AGENT_ARN="arn:aws:bedrock:us-west-2:123456789012:agent/ABCD1234" -
Auth0-specific環境変数を設定します (Auth0 にのみ必要):
export AUTH0_API_IDENTIFIER="your-api-identifier" -
クライアントを実行します。
python mcp_auth0_client.py
クライアントは自動的に以下を実行します。
-
URL で使用するエージェント ARN をエンコードする
-
MCP 呼び出しエンドポイント URL を作成する
-
認可リクエストに Auth0
audienceパラメータを追加する (Auth0 を使用する場合) -
OAuth 2.0 準拠の ID プロバイダーと連携する
付録
認証用に Cognito ユーザープールを設定する
新しいファイルを作成しsetup_cognito.sh、次のコンテンツを追加します。
#!/bin/bash # Create User Pool and capture Pool ID directly export POOL_ID=$(aws cognito-idp create-user-pool \ --pool-name "MyUserPool" \ --policies '{"PasswordPolicy":{"MinimumLength":8}}' \ --region $REGION | jq -r '.UserPool.Id') # Create App Client and capture Client ID directly export CLIENT_ID=$(aws cognito-idp create-user-pool-client \ --user-pool-id $POOL_ID \ --client-name "MyClient" \ --no-generate-secret \ --explicit-auth-flows "ALLOW_USER_PASSWORD_AUTH" "ALLOW_REFRESH_TOKEN_AUTH" \ --region $REGION | jq -r '.UserPoolClient.ClientId') # Create User aws cognito-idp admin-create-user \ --user-pool-id $POOL_ID \ --username $USERNAME \ --region $REGION \ --message-action SUPPRESS > /dev/null # Set Permanent Password aws cognito-idp admin-set-user-password \ --user-pool-id $POOL_ID \ --username $USERNAME \ --password $PASSWORD \ --region $REGION \ --permanent > /dev/null # Authenticate User and capture Access Token export BEARER_TOKEN=$(aws cognito-idp initiate-auth \ --client-id "$CLIENT_ID" \ --auth-flow USER_PASSWORD_AUTH \ --auth-parameters USERNAME=$USERNAME,PASSWORD=$PASSWORD \ --region $REGION | jq -r '.AuthenticationResult.AccessToken') # Output the required values echo "Pool id: $POOL_ID" echo "Discovery URL: https://cognito-idp.$REGION.amazonaws.com/$POOL_ID/.well-known/openid-configuration" echo "Client ID: $CLIENT_ID" echo "Bearer Token: $BEARER_TOKEN"
ターミナルウィンドウを開き、次の環境変数を設定します。
-
REGION– 使用する AWS リージョン -
USERNAME– 新しいユーザーのユーザー名 -
PASSWORD– 新しいユーザーのパスワード
export REGION=us-east-1 // set your desired Region export USERNAME=USER NAME export PASSWORD=PASSWORD
コマンド を使用してスクリプトを実行しますsource setup_cognito.sh。
注記
OAuth 認証のセットアップとサービスにリンクされたロールの詳細については、「インバウンド認証とアウトバウンド認証による認証と認可」を参照してください。
このスクリプトを実行したら、デプロイ設定で使用する次の値を書き留めます。
-
検出 URL:
agentcore createステップ中に使用されます -
クライアント ID:
agentcore createステップ中に使用されます -
ベアラートークン: デプロイされたサーバーを呼び出すときに使用されます
MCP インスペクターによるローカルテスト
MCP Inspector は、MCP サーバーをテストするためのビジュアルツールです。これを使用するには、以下が必要です。
-
Node.js と npm がインストールされている
MCP Inspector をインストールして実行します。
npx @modelcontextprotocol/inspector
これにより、次のようになります。
-
MCP Inspector サーバーを起動する
-
ターミナルに URL を表示する (通常は
http://localhost:6274)
Inspector を使用するには:
-
ブラウザ
http://localhost:6274で に移動する -
MCP サーバー URL ( )
http://localhost:8000/mcpを MCP Inspector 接続フィールドに貼り付けます。 -
サイドバーにツールが表示されます。
-
任意のツールをクリックしてテストする
-
パラメータを入力します (例:
add_numbersの場合は、aとbの値を入力します）。 -
「コールツール」をクリックして結果を表示します。
MCP インスペクターによるリモートテスト
MCP Inspector を使用してデプロイされたサーバーをテストすることもできます。まず、エージェント ARN を URL エンコードします。
export AGENT_ARN="arn:aws:bedrock-agentcore:us-west-2:123456789012:runtime/my_mcp_server-xyz123" echo -n $AGENT_ARN | jq -sRr '@uri'
これにより、URL エンコードされた ARN が出力されます。
arn%3Aaws%3Abedrock-agentcore%3Aus-west-2%3A123456789012%3Aruntime%2Fmy_mcp_server-xyz123
次に、MCP Inspector に接続します。
-
MCP Inspector を起動します。
npx @modelcontextprotocol/inspector -
ウェブインターフェイスで、次の操作を行います。
-
トランスポートとして「ストリーミング可能な HTTP」を選択する
-
エンコードされた ARN を使用してエージェントのエンドポイント URL を入力します。必ずエージェントの ARN と同じリージョンを使用してください。
https://bedrock-agentcore.REGION.amazonaws.com/runtimes/ENCODED_ARN/invocations?qualifier=DEFAULTus-west-2 の例:
https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/arn%3Aaws%3Abedrock-agentcore%3Aus-west-2%3A123456789012%3Aruntime%2Fmy_mcp_server-xyz123/invocations?qualifier=DEFAULT -
ヘッダー名
Authorizationと値を使用して、認証セクションにベアラートークンを追加するBearer YOUR_TOKEN -
「Connect」をクリックします。
-
-
ローカルで行ったのと同じようにツールをテストする
