Déployer des 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 obtenir un exemple, consultez https://github.com/awslabs/amazon-bedrock-agentcore-samples/tree/main/01-tutorials/01-AgentCore-runtime/02-hosting-MCP-server
Dans cette section, vous allez apprendre :
-
Comment créer un serveur MCP à l'aide d'outils
-
Comment tester votre serveur en local
-
Comment déployer votre serveur sur AWS
-
Comment appeler 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 Amazon Bedrock AgentCore Runtime 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 SDK de serveur MCP officiels.
Amazon Bedrock AgentCore prend en charge les serveurs MCP HTTP streamables avec ou sans état. Par défaut, le mode stateless (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 sans en-tête, 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 dynamique (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 demandes au cours d'un même appel. Pour plus d'informations, consultez la section Fonctionnalités du serveur Stateful MCP.
La charge utile de l'InvokeAgentRuntimeAPI est transmise directement, ce qui permet aux messages RPC de protocoles tels que MCP d'être facilement transmis par proxy.
Conditions préalables
-
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
-
FastMCP : crée un serveur MCP qui peut héberger vos outils
-
@mcp .tool () : Décorateur qui transforme vos fonctions Python en outils MCP
-
Outils : Trois outils simples illustrant différents types d'opérations
-
stateless_http=true : configure le serveur en mode stateless, qui est le mode par défaut pour les serveurs MCP de base
Astuce
Pour les serveurs MCP qui nécessitent des interactions multitours (élicitation) ou LLM-generated du contenu (échantillonnage), utilisez cette option stateless_http=False pour activer le mode dynamique. Les serveurs MCP Stateful maintiennent le contexte de session pour plusieurs requêtes dans le cadre d'une même invocation d'outil. Pour plus d'informations, consultez la section Fonctionnalités du serveur Stateful MCP.
É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.
Test avec le client MCP
À partir d'un nouveau terminal, créez un nouveau fichier my_mcp_client.py et exécutez-le en utilisant 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 la AgentCore CLI 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 être déployé dans 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é des charges de travail lors de l'utilisation de l'authentification OAuth. Pour obtenir des informations détaillées sur cette modification, consultez la section Rôle lié au service d'identité.
Après avoir configuré l'authentification, créez un nouveau projet avec le protocole MCP :
agentcore create --protocol MCP
Suivez les instructions interactives pour saisir le nom du projet. La CLI échafaude la structure du projet, y compris un fichier agentcore/agentcore.json de configuration. Copiez votre my_mcp_server.py fichier dans le répertoire du code agent du projet généré et assurez-vous que le point d'entrée agentcore/agentcore.json pointe vers le fichier de votre serveur.
Déployer vers AWS
Déployez votre agent :
agentcore deploy
Cette commande permettra de :
-
Package du code de votre agent et de 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 sur AWS
Après le déploiement, vous recevrez un ARN d'exécution de l'agent qui ressemble à ce qui suit :
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 (à distance)
Avant de tester, 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 porteur en tant que variable d'environnement :
export BEARER_TOKEN="bearer_token"
si vous transmettez un Accept en-tête, il doit respecter la norme MCPapplication/json ettext/event-stream.
Créez un nouveau fichier my_mcp_client_remote.py et exécutez-le en utilisant 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 respectent les normes d'authentification RFC 6749 (OAuth 2.0
401 Non autorisé - Authentification manquante
Lorsqu'aucun jeton porteur n'est fourni dans l'en-tête d'autorisation, 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 décrit 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), qui simplifie le processus de configuration du client en permettant aux clients de s'enregistrer 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 jusqu'à 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 by 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 sur l'enregistrement dynamique du client Auth0
Étape 5 : Création de votre projet pour le déploiement
Après avoir configuré l'authentification, créez un nouveau projet avec le protocole MCP :
agentcore create --protocol MCP
Suivez les instructions interactives pour saisir le nom du projet. La CLI échafaude la structure du projet, y compris un fichier agentcore/agentcore.json de configuration. Copiez votre my_mcp_server.py fichier dans le répertoire du code agent du projet généré et assurez-vous que le point d'entrée agentcore/agentcore.json pointe vers le fichier de votre serveur.
Étape 6 : Déployer vers AWS
Déployez votre agent :
agentcore deploy
Cette commande permettra de :
-
Package du code de votre agent et de 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 sur AWS
Après le déploiement, vous recevrez un ARN d'exécution de l'agent qui ressemble à ce qui suit :
arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_mcp_server-xyz123
Étape 7 : Invoquez votre agent déployé
Ce client est basé sur l'exemple officiel de client simple-auth-client du SDK MCP
Note
Lorsque vous utilisez Auth0 avec l'enregistrement dynamique des clients, 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 (cryptés) au lieu des jetons JWT standard. Le SDK MCP envoie le paramètre 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 la prioritéaudience. Pour plus d'informations, voir Communauté Auth0 - Jetons 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 des correctifs httpx pour injecter des User-Agent en-têtes dans toutes les requêtes HTTP. Cela est nécessaire car le SDK Python MCP 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 Python MCP
#!/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 :
-
Codez l'ARN de l'agent à utiliser dans l'URL
-
Construire 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
Configuration du 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 une configuration détaillée de l'authentification OAuth et des informations sur les Service-Linked rôles, voir Authentifier et autoriser avec l'authentification entrante et l'authentification sortante.
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 -
Token porteur : 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és
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'Inspector :
-
Accédez à
http://localhost:6274dans votre navigateur -
Collez l'URL du serveur MCP (
http://localhost:8000/mcp) dans le champ de connexion MCP Inspector -
Vous verrez vos outils 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 l'inspecteur MCP
Vous pouvez également tester votre serveur déployé à l'aide du MCP Inspector. 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 le MCP Inspector :
npx @modelcontextprotocol/inspector -
Dans l'interface Web :
-
Sélectionnez « Streamable HTTP » comme 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 « Connect »
-
-
Testez vos outils comme vous l'avez fait localement
