Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà.
Implementa server MCP in Runtime AgentCore
Amazon Bedrock AgentCore Runtime ti consente di distribuire ed eseguire server Model Context Protocol (MCP) nel Runtime. AgentCore Questa guida illustra come creare, testare e distribuire il tuo primo server MCP.
Per un esempio, consulta le nozioni di base sul server AgentCore MCP su. GitHub
In questa sezione, imparerai:
-
Come creare un server MCP con strumenti
-
Come testare il server localmente
-
Come distribuire il server su AWS
-
Come richiamare il server distribuito
Per ulteriori informazioni su MCP, vedere il contratto di protocollo MCP.
In che modo Amazon Bedrock supporta MCP AgentCore
Quando configuri Amazon Bedrock AgentCore Runtime con il protocollo MCP, il servizio prevede che i contenitori del server MCP siano disponibili nel percorso0.0.0.0:8000/mcp, che è il percorso predefinito supportato dalla maggior parte degli SDK ufficiali per server MCP.
Amazon Bedrock AgentCore supporta server MCP HTTP streamable sia stateless che stateful. Per impostazione predefinita, la modalità stateless () stateless_http=True è consigliata per i server MCP di base. La piattaforma aggiunge automaticamente un'Mcp-Session-Idintestazione per qualsiasi richiesta che ne sia sprovvista, in modo che i client MCP possano mantenere la continuità di connessione alla stessa sessione di Amazon Bedrock Runtime. AgentCore
Per i server MCP che richiedono interazioni a più turni (elicitazione), LLM-generated contenuti (campionamento) o notifiche di avanzamento, la modalità stateful () abilita queste funzionalità. stateless_http=False In modalità stateful, il runtime preserva lo stato della sessione MCP tra le richieste all'interno della stessa invocazione. Per ulteriori informazioni, consulta Funzionalità del server MCP Stateful.
Il payload dell'InvokeAgentRuntimeAPI viene trasmesso direttamente, consentendo di inviare facilmente tramite proxy i messaggi RPC di protocolli come MCP.
Prerequisiti
Prima di implementare un server MCP, assicurati di disporre di quanto segue:
-
Python 3.10 o versioni successive installato e conoscenza di base di Python
-
Un AWS account con le autorizzazioni appropriate e le credenziali locali configurate
Passaggio 1: crea il tuo server MCP
Installazione dei pacchetti obbligatori
Innanzitutto, installa il pacchetto MCP:
pip install mcp
Crea il tuo primo server MCP
Create un nuovo file chiamatomy_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")
Comprensione del codice
L'esempio utilizza i seguenti componenti:
-
FastMCP: crea un server MCP in grado di ospitare i tuoi strumenti
-
@mcp .tool (): Decoratore che trasforma le tue funzioni Python in strumenti MCP
-
Strumenti: tre semplici strumenti che dimostrano diversi tipi di operazioni
-
stateless_http=True: configura il server in modalità stateless, che è l'impostazione predefinita per i server MCP di base
Suggerimento
Per i server MCP che richiedono interazioni a più turni (elicitazione) o contenuti (campionamento), utilizzalo per abilitare la modalità stateful. LLM-generated stateless_http=False I server MCP con stato mantengono il contesto della sessione per più richieste all'interno della stessa chiamata dello strumento. Per ulteriori informazioni, consulta Funzionalità del server Steful MCP.
Passaggio 2: verifica il server MCP localmente
Avviate il vostro server MCP
Esegui il tuo server MCP localmente:
python my_mcp_server.py
Dovresti vedere un output che indica che il server è in esecuzione sulla porta8000.
Esegui il test con il client MCP
Da un nuovo terminale, crea un nuovo file my_mcp_client.py ed eseguilo usando 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())
È inoltre possibile testare il server utilizzando MCP Inspector come descritto in Test locali con MCP Inspector.
Fase 3: Distribuisci il tuo server MCP su AWS
Installa gli strumenti di distribuzione
Installa la AgentCore CLI:
npm install -g @aws/agentcore
Si utilizza la AgentCore CLI per distribuire l'agente su Runtime. AgentCore
Crea una cartella di progetto con la seguente struttura:
## 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
Crea un nuovo file chiamatorequirements.txt, aggiungi quanto segue:
mcp
requirements.txtspecifica i requisiti necessari all'agente per la distribuzione in Runtime. AgentCore
Crea il tuo progetto per la distribuzione
Prima di creare il progetto, devi configurare un pool di utenti Cognito per l'autenticazione, come descritto in Configurazione del pool di utenti Cognito per l'autenticazione. Ciò fornisce i token OAuth necessari per l'accesso sicuro al server distribuito.
Nota
A partire dal 7 ottobre 2025, Amazon Bedrock AgentCore utilizza un Service-Linked ruolo per le autorizzazioni di identità del carico di lavoro quando utilizza l'autenticazione OAuth. Per informazioni dettagliate su questa modifica, consulta Ruolo collegato al servizio Identity. Ruolo collegato al servizio di identità
Dopo aver impostato l'autenticazione, crea un nuovo progetto con il protocollo 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 crea una configurazione di runtime CUSTOM_JWT e impalca la struttura del progetto. I comandi copiano il server sul punto di app/MCPServer/main.py ingresso generato e aggiungono la sua dipendenza a. pyproject.toml
Distribuisci su AWS
Implementa il tuo agente:
agentcore deploy
Questo comando consentirà di:
-
Impacchettizza il codice e le dipendenze del tuo agente
-
Carica l'artefatto di distribuzione su Amazon S3
-
Crea un runtime Amazon Bedrock AgentCore
-
Distribuisci il tuo agente su AWS
Dopo la distribuzione, riceverai un ARN di runtime dell'agente simile a:
arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_mcp_server-xyz123
Fase 4: richiama il server MCP distribuito
Esegui il test con il client MCP (remoto)
Prima del test, imposta le seguenti variabili di ambiente:
-
L'agente di esportazione ARN come variabile di ambiente:
export AGENT_ARN="agent_arn" -
Token al portatore di esportazione come variabile di ambiente:
export BEARER_TOKEN="bearer_token"
se si Accept inserisce un'intestazione, questa deve seguire lo standard https://modelcontextprotocol.io/specification/2025-06-18/basic/transports#sending-messages-to-the-serverapplication/json e. text/event-stream
Crea un nuovo file my_mcp_client_remote.py ed eseguilo usando 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())
È inoltre possibile testare il server distribuito utilizzando MCP Inspector come descritto in Test remoto con MCP Inspector.
Risposte agli errori di autenticazione per gli agenti OAuth-Configured
OAuth-configured gli agenti seguono gli standard di autenticazione RFC 6749 (OAuth 2.0).
401 Non autorizzato - Autenticazione mancante
Quando non viene fornito alcun token Bearer nell'intestazione di autorizzazione, la risposta è:
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}"
Flusso end-to-end con Auth0
Questa sezione dimostra l'autenticazione OAuth utilizzando Auth0 come provider di identità. Utilizziamo Auth0 per questo esempio perché supporta Dynamic Client Registration (DCR), che semplifica il processo di configurazione del client consentendo ai client di registrarsi programmaticamente in fase di esecuzione.
Fase 1 - Fase 3: Create e testate il vostro server MCP
Segui i passaggi 1-3 dal passaggio 1: creazione del server MCP al passaggio 3: distribuzione del server MCP per creare e AWS testare il server MCP.
Fase 4: Creare l'applicazione Auth0
Segui le istruzioni di configurazione di Auth0 su Auth0 di Okta.
Abilita la registrazione dinamica del client:
-
Dashboard → Impostazioni → Avanzate
-
Attiva «Registrazione dinamica delle applicazioni OIDC» → ON
-
Salva le modifiche.
Per ulteriori informazioni, consulta la documentazione Auth0 Dynamic Client Registration.
Fase 5: Crea il tuo progetto per la distribuzione
Dopo aver impostato l'autenticazione, crea un nuovo progetto con il protocollo 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 ../..
Sostituisci i segnaposto Auth0 con i valori della tua applicazione Auth0. La CLI crea una configurazione di runtime CUSTOM_JWT e impalca la struttura del progetto. I comandi copiano il server sul punto di app/MCPServer/main.py ingresso generato e aggiungono la sua dipendenza a. pyproject.toml
Fase 6: Implementazione su AWS
Implementa il tuo agente:
agentcore deploy
Questo comando consentirà di:
-
Impacchettizza il codice e le dipendenze del tuo agente
-
Carica l'artefatto di distribuzione su Amazon S3
-
Crea un runtime Amazon Bedrock AgentCore
-
Distribuisci il tuo agente su AWS
Dopo la distribuzione, riceverai un ARN di runtime dell'agente simile a:
arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_mcp_server-xyz123
Passaggio 7: richiama l'agente distribuito
Questo client è basato sull'esempio ufficiale MCP SDK simple-auth-client con modifiche.
Nota
Quando si utilizza Auth0 con Dynamic Client Registration, è necessario includere il audience parametro nelle richieste di autorizzazione per ricevere i token JWT. Senza questo parametro, Auth0 restituisce token opachi o token JWE (crittografati) anziché token JWT standard. L'SDK MCP invia il parametro OAuth 2.0 (RFC 8707), ma Auth0 richiede il resource parametro OIDC per i token JWT. audience Entrambi i parametri hanno scopi audience simili, ma Auth0 dà la priorità. Per ulteriori informazioni, vedere Auth0 Community - Token JWT con registrazione dinamica delle applicazioni.
Crea un file denominato mcp_auth0_client.py con il codice seguente. Questo client gestisce Auth0-specific i requisiti, incluso il parametro audience:
Nota
Il codice include patch httpx per inserire User-Agent intestazioni in tutte le richieste HTTP. Ciò è necessario perché l'MCP Python SDK attualmente non include User-Agent intestazioni nelle sue richieste HTTP, il che può causare problemi con le regole WAF che richiedono intestazioni. AWS User-Agent Per ulteriori informazioni, consultate MCP Python SDK Issue #1664 e WAF managed rule groups.
#!/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()
Per utilizzare il client:
-
Imposta le variabili di ambiente richieste:
export AGENT_ARN="arn:aws:bedrock:us-west-2:123456789012:agent/ABCD1234" -
Imposta la variabile di Auth0-specific ambiente (richiesta solo per Auth0):
export AUTH0_API_IDENTIFIER="your-api-identifier" -
Esegui il client:
python mcp_auth0_client.py
Il client eseguirà automaticamente:
-
Codificare l'Agent ARN da utilizzare nell'URL
-
Costruisci l'URL dell'endpoint di chiamata MCP
-
Aggiungi il
audienceparametro Auth0 alle richieste di autorizzazione (quando usi Auth0) -
Lavora con qualsiasi provider di identità conforme a OAuth 2.0
Appendice
Configura il pool di utenti Cognito per l'autenticazione
Crea un nuovo file setup_cognito.sh e aggiungi i seguenti contenuti.
#!/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"
Apri una finestra di terminale e imposta le seguenti variabili di ambiente:
-
REGION— la AWS regione che vuoi usare -
USERNAME— il nome utente per il nuovo utente -
PASSWORD— la password per il nuovo utente
export REGION=us-east-1 # Set your desired Region export USERNAME="user-name" export PASSWORD="password"
Esegui lo script usando il comandosource setup_cognito.sh.
Nota
Per informazioni dettagliate sulla configurazione dell'autenticazione OAuth e sul Service-Linked ruolo, consulta Autenticazione e autorizzazione con Inbound Auth e Outbound Auth.
Dopo aver eseguito questo script, prendi nota dei seguenti valori da utilizzare nella configurazione di distribuzione:
-
URL di scoperta: utilizzato durante la
agentcore createfase -
ID cliente: utilizzato durante la
agentcore createfase -
Bearer Token: utilizzato quando si richiama il server distribuito
Test locali con MCP Inspector
MCP Inspector è uno strumento visivo per testare i server MCP. Per utilizzarlo, è necessario:
-
Node.js e npm è installato
Installa ed esegui MCP Inspector:
npx @modelcontextprotocol/inspector
Ciò consentirà di:
-
Avviare il server MCP Inspector
-
Visualizza un URL nel tuo terminale (in genere)
http://localhost:6274
Per utilizzare l'Inspector:
-
Accedi al
http://localhost:6274tuo browser -
Incolla l'URL del server MCP (
http://localhost:8000/mcp) nel campo di connessione MCP Inspector -
Vedrai i tuoi strumenti elencati nella barra laterale
-
Fai clic su uno strumento per testarlo
-
Inserisci i parametri (ad esempio, per
add_numbers, inserisci i valori peraeb) -
Fai clic su «Call Tool» per vedere il risultato
Test remoto con MCP Inspector
È inoltre possibile testare il server distribuito utilizzando MCP Inspector. Innanzitutto, il URL-encode tuo agente ARN:
export AGENT_ARN="arn:aws:bedrock-agentcore:us-west-2:123456789012:runtime/my_mcp_server-xyz123" echo -n $AGENT_ARN | jq -sRr '@uri'
Questo genera l'ARN: URL-encoded
arn%3Aaws%3Abedrock-agentcore%3Aus-west-2%3A123456789012%3Aruntime%2Fmy_mcp_server-xyz123
Quindi connettiti con MCP Inspector:
-
Avvia MCP Inspector:
npx @modelcontextprotocol/inspector -
Nell'interfaccia web:
-
Seleziona «Streamable HTTP» come trasporto
-
Inserisci l'URL dell'endpoint del tuo agente utilizzando l'ARN codificato. Assicurati di utilizzare la stessa regione dell'ARN del tuo agente:
https://bedrock-agentcore.REGION.amazonaws.com/runtimes/ENCODED_ARN/invocations?qualifier=DEFAULTEsempio per 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 -
Aggiungi il tuo token Bearer nella sezione Autenticazione con nome e valore dell'intestazione
AuthorizationBearer YOUR_TOKEN -
Fai clic su «Connetti»
-
-
Testa i tuoi strumenti proprio come hai fatto localmente
