Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.
Déploiement de serveurs MCP dans Runtime AgentCore
Amazon Bedrock AgentCore Runtime vous permet de déployer et d'exécuter des serveurs MCP (Model Context Protocol) dans le AgentCore Runtime. Ce guide explique comment créer, tester et déployer votre premier serveur MCP.
Pour un exemple, reportez-vous à la section Principes de base du serveur AgentCore MCP sur GitHub
Dans cette section, vous allez apprendre :
-
Comment créer un serveur MCP avec des outils
-
Comment tester votre serveur localement
-
Comment déployer votre serveur sur AWS
-
Comment invoquer votre serveur déployé
Pour plus d'informations sur le MCP, consultez le contrat de protocole MCP.
Comment Amazon Bedrock AgentCore soutient MCP
Lorsque vous configurez un AgentCore environnement d'exécution Amazon Bedrock avec le protocole MCP, le service s'attend à ce que les conteneurs du serveur MCP soient disponibles sur le chemin0.0.0.0:8000/mcp, qui est le chemin par défaut pris en charge par la plupart des kits SDK de serveurs MCP officiels.
Amazon Bedrock AgentCore prend en charge les serveurs MCP HTTP streamables avec état et avec état. Par défaut, le mode sans état (stateless_http=True) est recommandé pour les serveurs MCP de base. La plateforme ajoute automatiquement un Mcp-Session-Id en-tête pour toute demande qui n'en contient pas, afin que les clients MCP puissent maintenir la continuité de la connexion à la même session Amazon Bedrock AgentCore Runtime.
Pour les serveurs MCP qui nécessitent des interactions multitours (élicitation), LLM-generated du contenu (échantillonnage) ou des notifications de progression, le mode statique (stateless_http=False) active ces fonctionnalités. En mode dynamique, le moteur d'exécution préserve l'état de la session MCP pour toutes les requêtes d'une même invocation. Pour plus d'informations, consultez la section Fonctionnalités du serveur MCP Stateful.
La charge utile de l'InvokeAgentRuntimeAPI est transmise directement, ce qui permet de transmettre facilement par proxy les messages RPC de protocoles tels que MCP.
Conditions préalables
Avant de déployer un serveur MCP, assurez-vous de disposer des éléments suivants :
-
Python 3.10 ou supérieur installé et compréhension de base de Python
-
Un AWS compte avec les autorisations appropriées et les informations d'identification locales configurées
Étape 1 : Créez votre serveur MCP
Installation des packages obligatoires
Tout d'abord, installez le package MCP :
pip install mcp
Créez votre premier serveur MCP
Créez un nouveau fichier appelé 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")
Comprendre le code
L'exemple utilise les composants suivants :
-
FastMCP : crée un serveur MCP pouvant héberger vos outils
-
@mcp .tool () : décorateur qui transforme vos fonctions Python en outils MCP
-
Outils : trois outils simples qui illustrent différents types d'opérations
-
Stateless_http=True : configure le serveur en mode sans état, qui est le mode par défaut pour les serveurs MCP de base
Astuce
Pour les serveurs MCP qui nécessitent des interactions multi-tours (élicitation) ou LLM-generated du contenu (échantillonnage), utilisez stateless_http=False pour activer le mode dynamique. Les serveurs MCP dynamiques conservent le contexte de session pour plusieurs demandes au cours de la même invocation d'outil. Pour plus d'informations, consultez la section Fonctionnalités du serveur MCP Stateful.
Étape 2 : Testez votre serveur MCP localement
Démarrez votre serveur MCP
Exécutez votre serveur MCP localement :
python my_mcp_server.py
Vous devriez voir une sortie indiquant que le serveur fonctionne sur le port8000.
Testez avec le client MCP
À partir d'un nouveau terminal, créez un nouveau fichier my_mcp_client.py et exécutez-le à l'aide de 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())
Vous pouvez également tester votre serveur à l'aide de l'inspecteur MCP comme décrit dans la section Tests locaux avec l'inspecteur MCP.
Étape 3 : Déployez votre serveur MCP sur AWS
Installation des outils de déploiement
Installez la AgentCore CLI :
npm install -g @aws/agentcore
Vous utilisez l' AgentCore interface de ligne de commande pour déployer votre agent sur AgentCore Runtime.
Créez un dossier de projet avec la structure suivante :
## 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
Créez un nouveau fichier appelérequirements.txt, ajoutez-y ce qui suit :
mcp
requirements.txtspécifie les exigences dont l'agent a besoin pour le déploiement sur AgentCore Runtime.
Créez votre projet pour le déploiement
Avant de créer votre projet, vous devez configurer un groupe d'utilisateurs Cognito pour l'authentification, comme décrit dans Configuration du groupe d'utilisateurs Cognito pour l'authentification. Cela fournit les jetons OAuth nécessaires pour un accès sécurisé à votre serveur déployé.
Note
À compter du 7 octobre 2025, Amazon Bedrock AgentCore utilise un Service-Linked rôle pour les autorisations d'identité de la charge de travail lors de l'utilisation de l'authentification OAuth. Pour plus d'informations sur cette modification, consultez la section Rôle lié au service d'identité.
Après avoir configuré l'authentification, élafaudez un nouveau projet avec le protocole MCP :
agentcore create --project-name MCPServerProject --no-agent cd MCPServerProject agentcore add agent \ --name MCPServer \ --language Python \ --protocol MCP \ --authorizer-type CUSTOM_JWT \ --discovery-url "https://cognito-idp.$REGION.amazonaws.com/$POOL_ID/.well-known/openid-configuration" \ --allowed-clients "$CLIENT_ID" \ --request-header-allowlist Authorization cp ../my_mcp_server.py app/MCPServer/main.py cd app/MCPServer uv add mcp cd ../..
La CLI crée une configuration d'exécution CUSTOM_JWT et échafaude la structure du projet. Les commandes copient votre serveur sur le app/MCPServer/main.py point d'entrée généré et y ajoutent sa dépendance. pyproject.toml
Déployez vers AWS
Déployez votre agent :
agentcore deploy
Cette commande va :
-
Regroupez le code de votre agent et ses dépendances
-
Téléchargez l'artefact de déploiement sur Amazon S3
-
Création d'un environnement d'exécution Amazon Bedrock AgentCore
-
Déployez votre agent pour AWS
Après le déploiement, vous recevrez un ARN d'exécution de l'agent qui ressemble à :
arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_mcp_server-xyz123
Étape 4 : Invoquez votre serveur MCP déployé
Test avec le client MCP (distant)
Avant le test, définissez les variables d'environnement suivantes :
-
Exportez l'ARN de l'agent en tant que variable d'environnement :
export AGENT_ARN="agent_arn" -
Exportez le jeton du porteur en tant que variable d'environnement :
export BEARER_TOKEN="bearer_token"
si vous transmettez un Accept en-tête, celui-ci doit respecter la norme application/json ettext/event-stream.
Créez un nouveau fichier my_mcp_client_remote.py et exécutez-le à l'aide de 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())
Vous pouvez également tester votre serveur déployé à l'aide de l'inspecteur MCP, comme décrit dans Tests à distance avec l'inspecteur MCP.
Réponses aux erreurs d'authentification pour les OAuth-Configured agents
OAuth-configured les agents suivent les normes d'authentification RFC 6749 (OAuth 2.0).
401 Non autorisé - Authentification manquante
Lorsqu'aucun jeton Bearer n'est fourni dans l'en-tête Authorization, la réponse est la suivante :
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}"
Flux de bout en bout avec Auth0
Cette section présente l'authentification OAuth en utilisant Auth0 comme fournisseur d'identité. Nous utilisons Auth0 pour cet exemple car il prend en charge l'enregistrement dynamique des clients (DCR), ce qui simplifie le processus de configuration du client en permettant aux clients de s'enregistrer eux-mêmes par programmation lors de l'exécution.
Étape 1 - Étape 3 : Création et test de votre serveur MCP
Suivez les étapes 1 à 3 de l'étape 1 : Créez votre serveur MCP à l'étape 3 : déployez votre serveur MCP AWS pour créer et tester votre serveur MCP.
Étape 4 : Création de l'application Auth0
Suivez les instructions de configuration d'Auth0 sur Auth0 d'Okta.
Activez l'enregistrement dynamique des clients :
-
Tableau de bord → Paramètres → Avancé
-
Activez « Enregistrement dynamique des applications OIDC » → Activé
-
Enregistrer les modifications.
Pour plus d'informations, consultez la documentation https://auth0.com/docs/get-started/applications/dynamic-client-registration
Étape 5 : Création de votre projet pour le déploiement
Après avoir configuré l'authentification, élafaudez un nouveau projet avec le protocole MCP :
agentcore create --project-name MCPServerProject --no-agent cd MCPServerProject agentcore add agent \ --name MCPServer \ --language Python \ --protocol MCP \ --authorizer-type CUSTOM_JWT \ --discovery-url "<AUTH0_DISCOVERY_URL>" \ --allowed-clients "<AUTH0_CLIENT_ID>" \ --request-header-allowlist Authorization cp ../my_mcp_server.py app/MCPServer/main.py cd app/MCPServer uv add mcp cd ../..
Remplacez les espaces réservés Auth0 par des valeurs provenant de votre application Auth0. La CLI crée une configuration d'exécution CUSTOM_JWT et échafaude la structure du projet. Les commandes copient votre serveur sur le app/MCPServer/main.py point d'entrée généré et y ajoutent sa dépendance. pyproject.toml
Étape 6 : Déploiement sur AWS
Déployez votre agent :
agentcore deploy
Cette commande va :
-
Regroupez le code de votre agent et ses dépendances
-
Téléchargez l'artefact de déploiement sur Amazon S3
-
Création d'un environnement d'exécution Amazon Bedrock AgentCore
-
Déployez votre agent pour AWS
Après le déploiement, vous recevrez un ARN d'exécution de l'agent qui ressemble à :
arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_mcp_server-xyz123
Étape 7 : Invoquer votre agent déployé
Ce client est basé sur l'exemple de client simple-auth-client officiel du SDK MCP avec des modifications.
Note
Lorsque vous utilisez Auth0 avec Dynamic Client Registration, vous devez inclure le audience paramètre dans les demandes d'autorisation pour recevoir des jetons JWT. Sans ce paramètre, Auth0 renvoie des jetons opaques ou des jetons JWE (chiffrés) au lieu des jetons JWT standard. Le SDK MCP envoie le paramètre d'OAuth 2.0 (RFC 8707), mais Auth0 nécessite le resource paramètre OIDC pour les jetons JWT. audience Les deux paramètres ont des objectifs similaires, mais Auth0 donne des prioritésaudience. Pour plus d'informations, consultez Auth0 Community - Tokens JWT avec enregistrement dynamique des applications.
Créez un fichier nommé mcp_auth0_client.py avec le code suivant. Ce client gère les Auth0-specific exigences, y compris le paramètre d'audience :
Note
Le code inclut un correctif httpx pour injecter des User-Agent en-têtes dans toutes les requêtes HTTP. Cela est nécessaire car le SDK MCP Python n'inclut actuellement pas d' User-Agent en-têtes dans ses requêtes HTTP, ce qui peut entraîner des problèmes avec les règles AWS WAF qui nécessitent des en-têtes. User-Agent Pour plus d'informations, consultez le numéro #1664 du SDK MCP Python
#!/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()
Pour utiliser le client :
-
Définissez les variables d'environnement requises :
export AGENT_ARN="arn:aws:bedrock:us-west-2:123456789012:agent/ABCD1234" -
Définissez la variable d' Auth0-specific environnement (obligatoire uniquement pour Auth0) :
export AUTH0_API_IDENTIFIER="your-api-identifier" -
Exécutez le client :
python mcp_auth0_client.py
Le client va automatiquement :
-
Encodez l'ARN de l'agent à utiliser dans l'URL
-
Construisez l'URL du point de terminaison d'invocation MCP
-
Ajouter le
audienceparamètre Auth0 aux demandes d'autorisation (lors de l'utilisation de Auth0) -
Travaillez avec n'importe quel fournisseur d'identité conforme à OAuth 2.0
Annexe
Configurer le groupe d'utilisateurs Cognito pour l'authentification
Créez un nouveau fichier setup_cognito.sh et ajoutez le contenu suivant.
#!/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"
Ouvrez une fenêtre de terminal et définissez les variables d'environnement suivantes :
-
REGION— la AWS région que vous souhaitez utiliser -
USERNAME— le nom d'utilisateur du nouvel utilisateur -
PASSWORD— le mot de passe du nouvel utilisateur
export REGION=us-east-1 # Set your desired Region export USERNAME="user-name" export PASSWORD="password"
Exécutez le script à l'aide de la commandesource setup_cognito.sh.
Note
Pour obtenir des informations détaillées sur la configuration de l'authentification OAuth et les informations sur les Service-Linked rôles, voir Authentification et autorisation avec Inbound Auth et Outbound Auth.
Après avoir exécuté ce script, notez les valeurs suivantes à utiliser dans la configuration de déploiement :
-
URL de découverte : utilisée lors de l'
agentcore createétape -
ID client : utilisé lors de l'
agentcore createétape -
Bearer Token : utilisé lors de l'appel de votre serveur déployé
Tests locaux avec l'inspecteur MCP
Le MCP Inspector est un outil visuel permettant de tester les serveurs MCP. Pour l'utiliser, il vous faut :
-
Node.js et npm installé
Installez et exécutez le MCP Inspector :
npx @modelcontextprotocol/inspector
Cela permettra de :
-
Démarrez le serveur MCP Inspector
-
Afficher une URL dans votre terminal (généralement
http://localhost:6274)
Pour utiliser l'inspecteur, procédez comme suit :
-
Naviguez jusqu'
http://localhost:6274à dans votre navigateur -
Collez l'URL du serveur MCP (
http://localhost:8000/mcp) dans le champ de connexion MCP Inspector -
Vos outils seront répertoriés dans la barre latérale
-
Cliquez sur n'importe quel outil pour le tester
-
Renseignez les paramètres (par exemple, pour
add_numbers, entrez des valeurs pouraetb) -
Cliquez sur « Outil d'appel » pour voir le résultat
Tests à distance avec MCP Inspector
Vous pouvez également tester votre serveur déployé à l'aide de l'inspecteur MCP. Tout d'abord, l'ARN de URL-encode votre agent :
export AGENT_ARN="arn:aws:bedrock-agentcore:us-west-2:123456789012:runtime/my_mcp_server-xyz123" echo -n $AGENT_ARN | jq -sRr '@uri'
Cela génère l' URL-encoded ARN :
arn%3Aaws%3Abedrock-agentcore%3Aus-west-2%3A123456789012%3Aruntime%2Fmy_mcp_server-xyz123
Connectez-vous ensuite au MCP Inspector :
-
Démarrez l'inspecteur MCP :
npx @modelcontextprotocol/inspector -
Dans l'interface Web :
-
Sélectionnez « Streamable HTTP » comme mode de transport
-
Entrez l'URL du point de terminaison de votre agent à l'aide de l'ARN codé. Assurez-vous d'utiliser la même région que l'ARN de votre agent :
https://bedrock-agentcore.REGION.amazonaws.com/runtimes/ENCODED_ARN/invocations?qualifier=DEFAULTExemple pour 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 -
Ajoutez votre jeton Bearer dans la section Authentification avec le nom
Authorizationet la valeur de l'en-têteBearer YOUR_TOKEN -
Cliquez sur « Connecter »
-
-
Testez vos outils comme vous l'avez fait localement
