# Bereitstellen von MCP-Servern in Runtime AgentCore Mit Amazon Bedrock AgentCore Runtime können Sie Model Context Protocol (MCP) -Server in der AgentCore Runtime bereitstellen und ausführen. Dieser Leitfaden führt Sie durch die Erstellung, das Testen und die Bereitstellung Ihres ersten MCP-Servers. Ein Beispiel finden Sie unter [https://github.com/awslabs/amazon-bedrock-agentcore-samples/tree/main/01-tutorials/01-AgentCore-runtime/02-hosting-MCP-server](https://github.com/awslabs/amazon-bedrock-agentcore-samples/tree/main/01-tutorials/01-AgentCore-runtime/02-hosting-MCP-server). In diesem Abschnitt lernen Sie: + Wie erstellt man einen MCP-Server mit Tools + Wie testest du deinen Server lokal + Wie stellen Sie Ihren Server bereit AWS + Wie rufen Sie Ihren bereitgestellten Server auf Weitere Informationen zu MCP finden Sie unter [MCP-Protokollvertrag](runtime-mcp-protocol-contract.md). **Topics** + [So AgentCore unterstützt Amazon Bedrock MCP](#runtime-mcp-how-it-works) + [Voraussetzungen](#runtime-mcp-prerequisites) + [Schritt 1: Erstellen Sie Ihren MCP-Server](#runtime-mcp-create-server) + [Schritt 2: Testen Sie Ihren MCP-Server lokal](#runtime-mcp-test-locally) + [Schritt 3: Stellen Sie Ihren MCP-Server bereit AWS](#runtime-mcp-deploy-aws) + [Schritt 4: Rufen Sie Ihren bereitgestellten MCP-Server auf](#runtime-mcp-invoke-server) + [Antworten auf Authentifizierungsfehler für Agenten OAuth-Configured](#runtime-mcp-auth-error-responses) + [Ende-zu-Ende-Fluss mit Auth0](#runtime-mcp-auth0-flow) + [Anhang](#runtime-mcp-appendix) ## So AgentCore unterstützt Amazon Bedrock MCP Wenn Sie eine Amazon Bedrock AgentCore Runtime mit dem MCP-Protokoll konfigurieren, erwartet der Service, dass MCP-Servercontainer unter dem Pfad verfügbar sind. Dies ist der Standardpfad`0.0.0.0:8000/mcp`, der von den meisten offiziellen MCP-Server-SDKs unterstützt wird. Amazon Bedrock AgentCore unterstützt sowohl statuslose als auch statusbehaftete Streamable-HTTP-MCP-Server. Standardmäßig wird der Stateless Mode () `stateless_http=True` für einfache MCP-Server empfohlen. Die Plattform fügt automatisch einen `Mcp-Session-Id` Header für jede Anfrage hinzu, ohne einen, sodass MCP-Clients die Verbindungskontinuität zu derselben Amazon Bedrock AgentCore Runtime-Sitzung aufrechterhalten können. Für MCP-Server, die Multi-Turn-Interaktionen (Auslösung), LLM-generated Inhalte (Sampling) oder Fortschrittsbenachrichtigungen benötigen, aktiviert der Stateful-Modus () diese Funktionen. `stateless_http=False` Im Stateful-Modus behält die Runtime den MCP-Sitzungsstatus für alle Anfragen innerhalb desselben Aufrufs bei. Weitere Informationen finden Sie unter [Stateful-MCP-Serverfunktionen](mcp-stateful-features.md). Die Nutzdaten der [InvokeAgentRuntime](https://docs.aws.amazon.com/bedrock-agentcore/latest/APIReference/API_InvokeAgentRuntime.html)API werden direkt weitergeleitet, sodass RPC-Nachrichten von Protokollen wie MCP problemlos per Proxy weitergeleitet werden können. ## Voraussetzungen + Python 3.10 oder höher installiert und Grundkenntnisse von Python + Ein AWS Konto mit entsprechenden Berechtigungen und konfigurierten lokalen Anmeldeinformationen ## Schritt 1: Erstellen Sie Ihren MCP-Server ### Installieren erforderlicher Pakete Installieren Sie zunächst das MCP-Paket: ``` pip install mcp ``` ### Erstellen Sie Ihren ersten MCP-Server Erstellen Sie eine neue Datei mit dem Namen`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") ``` ### Den Code verstehen + **FastMCP**: Erstellt einen MCP-Server, auf dem Ihre Tools gehostet werden können + **@mcp .tool ()**: Decorator, der Ihre Python-Funktionen in MCP-Tools umwandelt + **Tools**: Drei einfache Tools, die verschiedene Arten von Operationen demonstrieren + **stateless\_http=True**: Konfiguriert den Server im statenlosen Modus, der Standardeinstellung für einfache MCP-Server **Tipp** Verwenden Sie für MCP-Server, die Multi-Turn-Interaktionen (Elicitation) oder Content (Sampling) erfordern, um den Stateful-Modus zu aktivieren. LLM-generated `stateless_http=False` Stateful-MCP-Server behalten den Sitzungskontext für mehrere Anfragen innerhalb desselben Toolaufrufs bei. Weitere Informationen finden Sie unter [Stateful](mcp-stateful-features.md) MCP-Serverfunktionen. ## Schritt 2: Testen Sie Ihren MCP-Server lokal ### Starten Sie Ihren MCP-Server Führen Sie Ihren MCP-Server lokal aus: ``` python my_mcp_server.py ``` Sie sollten eine Ausgabe sehen, die darauf hinweist, dass der Server auf dem Port `8000` läuft. ### Testen Sie mit dem MCP-Client Erstellen Sie von einem neuen Terminal aus eine neue Datei `my_mcp_client.py` und führen Sie sie mit `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()) ``` Sie können Ihren Server auch mit dem MCP Inspector testen, wie unter [Lokales Testen mit MCP](#runtime-mcp-appendix-b) Inspector beschrieben. ## Schritt 3: Stellen Sie Ihren MCP-Server bereit AWS ### Installieren Sie die Bereitstellungstools Installieren Sie die AgentCore CLI: ``` npm install -g @aws/agentcore ``` Sie verwenden die AgentCore CLI, um Ihren Agenten in AgentCore Runtime bereitzustellen. Erstellen Sie einen Projektordner mit der folgenden Struktur: ``` ## 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 ``` Erstellen Sie eine neue Datei mit dem Namen `requirements.txt` und fügen Sie ihr Folgendes hinzu: ``` mcp ``` `requirements.txt`gibt die Anforderungen an, die der Agent für die Bereitstellung in AgentCore Runtime benötigt. ### Erstellen Sie Ihr Projekt für die Bereitstellung Bevor Sie Ihr Projekt erstellen, müssen Sie einen Cognito-Benutzerpool für die Authentifizierung einrichten, wie unter [Cognito-Benutzerpool für die Authentifizierung einrichten](#runtime-mcp-appendix-a) beschrieben. Dadurch werden die OAuth-Token bereitgestellt, die für den sicheren Zugriff auf Ihren bereitgestellten Server erforderlich sind. **Anmerkung** Ab dem **7. Oktober 2025** AgentCore verwendet Amazon Bedrock eine Service-Linked Rolle für Workload-Identitätsberechtigungen, wenn die OAuth-Authentifizierung verwendet wird. Ausführliche Informationen zu dieser Änderung finden Sie unter Rolle, die mit dem [Identitätsdienst](service-linked-roles.md#identity-service-linked-role) verknüpft ist. Nachdem Sie die Authentifizierung eingerichtet haben, erstellen Sie ein Gerüst für ein neues Projekt mit dem MCP-Protokoll: ``` agentcore create --protocol MCP ``` Folgen Sie den interaktiven Anweisungen, um einen Projektnamen einzugeben. Die CLI bildet das Gerüst für die Projektstruktur einschließlich einer `agentcore/agentcore.json` Konfigurationsdatei. Kopieren Sie Ihre `my_mcp_server.py` Datei in das Agentencode-Verzeichnis des generierten Projekts und stellen Sie sicher, dass der Eingangspunkt auf Ihre Serverdatei `agentcore/agentcore.json` verweist. ### Bereitstellen auf AWS Stellen Sie Ihren Agenten bereit: ``` agentcore deploy ``` Dieser Befehl wird: 1. Verpacken Sie Ihren Agentencode und Ihre Abhängigkeiten 1. Laden Sie das Bereitstellungsartefakt auf Amazon S3 hoch 1. Erstellen Sie eine Amazon Bedrock-Runtime AgentCore 1. Stellen Sie Ihren Agenten bereit für AWS Nach der Bereitstellung erhalten Sie einen Runtime-ARN für Agenten, der wie folgt aussieht: ``` arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_mcp_server-xyz123 ``` ## Schritt 4: Rufen Sie Ihren bereitgestellten MCP-Server auf ### Testen Sie mit dem MCP-Client (remote) Stellen Sie vor dem Testen die folgenden Umgebungsvariablen ein: + Agenten-ARN als Umgebungsvariable exportieren: `export AGENT_ARN="agent_arn"` + Bearer-Token als Umgebungsvariable exportieren: `export BEARER_TOKEN="bearer_token"` Wenn Sie einen `Accept` Header übergeben, muss dieser dem [MCP-Standard](https://modelcontextprotocol.io/specification/2025-06-18/basic/transports#sending-messages-to-the-server) entsprechen. Zulässige Medientypen sind `application/json` und`text/event-stream`. Erstellen Sie eine neue Datei `my_mcp_client_remote.py` und führen Sie sie mit `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()) ``` Sie können Ihren bereitgestellten Server auch mit dem MCP Inspector testen, wie unter [Remote Testing with MCP](#runtime-mcp-appendix-c) Inspector beschrieben. ## Antworten auf Authentifizierungsfehler für Agenten OAuth-Configured OAuth-configured Agenten folgen den [Authentifizierungsstandards RFC 6749 (OAuth 2.0](https://datatracker.ietf.org/doc/html/rfc6749)). Fehlt die Authentifizierung, gibt der Dienst eine Antwort 401 Unauthorized mit einem WWW-Authenticate Header (gemäß [RFC 7235](https://datatracker.ietf.org/doc/html/rfc7235)) zurück, sodass Clients die Endpunkte des Autorisierungsservers über die API ermitteln können. GetRuntimeProtectedResourceMetadata ### 401 Nicht autorisiert — Fehlende Authentifizierung Wenn im Authorization-Header kein Bearer-Token angegeben ist, lautet die Antwort: ``` 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}" ``` ## Ende-zu-Ende-Fluss mit Auth0 In diesem Abschnitt wird die OAuth-Authentifizierung mit Auth0 als Identitätsanbieter demonstriert. Wir verwenden Auth0 für dieses Beispiel, weil es **Dynamic Client Registration (DCR)** unterstützt, was den Client-Einrichtungsprozess vereinfacht, da sich Clients zur Laufzeit programmgesteuert registrieren können. ### Schritt 1 — Schritt 3: Erstellen und testen Sie Ihren MCP-Server Folgen Sie den Schritten 1—3 von [Schritt 1: MCP-Server erstellen](#runtime-mcp-create-server) bis [Schritt 3: Bereitstellen Ihres MCP-Servers auf, um Ihren MCP-Server AWS zu](#runtime-mcp-deploy-aws) erstellen und zu testen. ### Schritt 4: Auth0-Anwendung erstellen Folgen Sie den Anweisungen zur Einrichtung von Auth0 unter [Auth0](identity-idp-auth0.md) von Okta. **Dynamische Client-Registrierung aktivieren:** 1. Dashboard → Einstellungen → Erweitert 1. Schalten Sie „Dynamische OIDC-Anwendungsregistrierung“ → EIN 1. Speichern der Änderungen Weitere Informationen finden Sie in der Dokumentation zur [Auth0 Dynamic](https://auth0.com/docs/get-started/applications/dynamic-client-registration) Client Registration. ### Schritt 5: Erstellen Sie Ihr Projekt für die Bereitstellung Nachdem Sie die Authentifizierung eingerichtet haben, erstellen Sie ein neues Projekt mit dem MCP-Protokoll: ``` agentcore create --protocol MCP ``` Folgen Sie den interaktiven Anweisungen, um einen Projektnamen einzugeben. Die CLI bildet das Gerüst für die Projektstruktur einschließlich einer `agentcore/agentcore.json` Konfigurationsdatei. Kopieren Sie Ihre `my_mcp_server.py` Datei in das Agentencode-Verzeichnis des generierten Projekts und stellen Sie sicher, dass der Eingangspunkt auf Ihre Serverdatei `agentcore/agentcore.json` verweist. ### Schritt 6: Bereitstellen auf AWS Stellen Sie Ihren Agenten bereit: ``` agentcore deploy ``` Dieser Befehl wird: + Verpacken Sie Ihren Agentencode und Ihre Abhängigkeiten + Laden Sie das Bereitstellungsartefakt auf Amazon S3 hoch + Erstellen Sie eine Amazon Bedrock-Runtime AgentCore + Stellen Sie Ihren Agenten bereit für AWS Nach der Bereitstellung erhalten Sie einen Runtime-ARN für Agenten, der wie folgt aussieht: ``` arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_mcp_server-xyz123 ``` ### Schritt 7: Rufen Sie Ihren bereitgestellten Agenten auf Dieser Client basiert auf dem [offiziellen MCP SDK simple-auth-client-Beispiel](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/clients/simple-auth-client/mcp_simple_auth_client/main.py) mit Änderungen. Auth0-specific **Anmerkung** Wenn Sie Auth0 mit dynamischer Client-Registrierung verwenden, müssen Sie den `audience` Parameter in Autorisierungsanfragen angeben, um JWT-Token zu erhalten. Ohne diesen Parameter gibt Auth0 undurchsichtige Token oder (verschlüsselte) JWE-Token anstelle von Standard-JWT-Token zurück. Das MCP-SDK sendet den Parameter von OAuth 2.0 (RFC 8707), aber Auth0 benötigt den `resource` OIDC-Parameter für JWT-Token. `audience` Beide Parameter `audience` dienen ähnlichen Zwecken, aber Auth0 priorisiert. Weitere Informationen finden Sie unter [Auth0 Community — JWT-Token](https://community.auth0.com/t/jwt-tokens-with-dynamic-application-registration/189741) mit dynamischer Anwendungsregistrierung. Erstellen Sie eine Datei mit dem Namen `mcp_auth0_client.py` und dem folgenden Code. Dieser Client verarbeitet Auth0-specific Anforderungen wie den Zielgruppenparameter: **Anmerkung** Der Code beinhaltet HTTPx-Patching, um User-Agent Header in alle HTTP-Anfragen einzufügen. Dies ist notwendig, da das MCP-Python-SDK derzeit keine User-Agent Header in seine HTTP-Anfragen einbezieht, was zu Problemen mit AWS WAF-Regeln führen kann, die Header erfordern. User-Agent Weitere Informationen finden Sie unter [MCP Python SDK Issue \#1664](https://github.com/modelcontextprotocol/python-sdk/issues/1664) und [AWS WAF-verwaltete Regelgruppen](https://docs.aws.amazon.com/waf/latest/developerguide/aws-managed-rule-groups-baseline.html). #### MCP Python Auth0-Clientcode ``` #!/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"""

Authorization Successful!

You can close this window and return to the terminal.

""") 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"""

Authorization Failed

Error: {query_params["error"][0]}

You can close this window and return to the terminal.

""".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 [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 ', 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() ``` Um den Client zu verwenden: 1. Stellen Sie die erforderlichen Umgebungsvariablen ein: ``` export AGENT_ARN="arn:aws:bedrock:us-west-2:123456789012:agent/ABCD1234" ``` 1. Auth0-specific Umgebungsvariable setzen (nur für Auth0 erforderlich): ``` export AUTH0_API_IDENTIFIER="your-api-identifier" ``` 1. Führen Sie den Client aus: ``` python mcp_auth0_client.py ``` Der Client wird automatisch: + Codieren Sie den Agenten-ARN für die Verwendung in der URL + Konstruieren Sie die URL des MCP-Aufruf-Endpunkts + Fügen Sie den `audience` Auth0-Parameter zu Autorisierungsanfragen hinzu (wenn Sie Auth0 verwenden) + Arbeiten Sie mit jedem OAuth 2.0-kompatiblen Identitätsanbieter ## Anhang ### Cognito-Benutzerpool für die Authentifizierung einrichten Erstellen Sie eine neue Datei `setup_cognito.sh` und fügen Sie den folgenden Inhalt hinzu. ``` #!/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" ``` Öffnen Sie ein Terminalfenster und legen Sie die folgenden Umgebungsvariablen fest: + {{REGION}}— die AWS Region, die Sie verwenden möchten + {{USERNAME}}— der Benutzername für den neuen Benutzer + {{PASSWORD}}— das Passwort für den neuen Benutzer ``` export REGION=us-east-1 // set your desired Region export USERNAME=USER NAME export PASSWORD=PASSWORD ``` Führen Sie das Skript mit dem Befehl aus`source setup_cognito.sh`. **Anmerkung** Ausführliche Informationen zur Einrichtung der OAuth-Authentifizierung und zu den Service-Linked Rollen finden Sie unter [Authentifizieren und Autorisieren mit Inbound Auth und Outbound](runtime-oauth.md) Auth. Notieren Sie sich nach der Ausführung dieses Skripts die folgenden Werte für die Verwendung in der Bereitstellungskonfiguration: + Discovery-URL: Wird während des `agentcore create` Schritts verwendet + Client-ID: Wird während des `agentcore create` Schritts verwendet + Bearer-Token: Wird beim Aufrufen Ihres bereitgestellten Servers verwendet ### Lokales Testen mit MCP Inspector Der MCP Inspector ist ein visuelles Tool zum Testen von MCP-Servern. Um es zu verwenden, benötigen Sie: + Node.js und npm installiert Installieren und starten Sie den MCP Inspector: ``` npx @modelcontextprotocol/inspector ``` Das wird: + Starten Sie den MCP Inspector Server + Zeigen Sie eine URL in Ihrem Terminal an (normalerweise`http://localhost:6274`) So verwenden Sie den Inspector: 1. Navigieren Sie `http://localhost:6274` in Ihrem Browser zu 1. Fügen Sie die MCP-Server-URL (`http://localhost:8000/mcp`) in das MCP Inspector-Verbindungsfeld ein 1. Ihre Tools werden in der Seitenleiste aufgelistet 1. Klicken Sie auf ein Tool, um es zu testen 1. Füllen Sie die Parameter aus (z. B. für`add_numbers`, geben Sie Werte für `a` und ein`b`) 1. Klicken Sie auf „Call Tool“, um das Ergebnis zu sehen ### Ferntests mit MCP Inspector Sie können Ihren bereitgestellten Server auch mit dem MCP Inspector testen. Zuerst URL-encode Ihr Agent ARN: ``` export AGENT_ARN="arn:aws:bedrock-agentcore:us-west-2:123456789012:runtime/my_mcp_server-xyz123" echo -n $AGENT_ARN | jq -sRr '@uri' ``` Dies gibt den URL-encoded ARN aus: ``` arn%3Aaws%3Abedrock-agentcore%3Aus-west-2%3A123456789012%3Aruntime%2Fmy_mcp_server-xyz123 ``` Stellen Sie dann eine Verbindung zum MCP Inspector her: 1. Starten Sie den MCP Inspector: ``` npx @modelcontextprotocol/inspector ``` 1. In der Weboberfläche: + Wählen Sie „Streamable HTTP“ als Transport + Geben Sie die Endpunkt-URL Ihres Agenten mit dem codierten ARN ein. Stellen Sie sicher, dass Sie dieselbe Region wie den ARN Ihres Agenten verwenden: ``` https://bedrock-agentcore.REGION.amazonaws.com/runtimes/ENCODED_ARN/invocations?qualifier=DEFAULT ``` Beispiel für us-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 ``` + Fügen Sie Ihr Bearer-Token im Abschnitt Authentifizierung mit Header-Namen und Wert hinzu `Authorization` `Bearer YOUR_TOKEN` + Klicken Sie auf „Connect“ 1. Testen Sie Ihre Tools genauso, wie Sie es vor Ort getan haben