Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés.
Implemente servidores MCP en AgentCore tiempo de ejecución
Amazon Bedrock AgentCore Runtime le permite implementar y ejecutar servidores de Model Context Protocol (MCP) en tiempo de ejecución. AgentCore Esta guía le explica cómo crear, probar e implementar su primer servidor MCP.
Para ver un ejemplo, consulte los conceptos básicos del servidor AgentCore MCP en. GitHub
En esta sección, aprenderá lo siguiente:
-
Cómo crear un servidor MCP con herramientas
-
Cómo probar tu servidor localmente
-
Cómo implementar su servidor en AWS
-
Cómo invocar el servidor desplegado
Para obtener más información sobre MCP, consulte el contrato de protocolo MCP.
¿Cómo apoya Amazon Bedrock AgentCore a MCP?
Al configurar un Amazon Bedrock AgentCore Runtime con el protocolo MCP, el servicio espera que los contenedores del servidor MCP estén disponibles en la ruta0.0.0.0:8000/mcp, que es la ruta predeterminada que admiten la mayoría de los SDK oficiales de los servidores MCP.
Amazon Bedrock AgentCore admite servidores MCP HTTP transmisibles tanto sin estado como con estado. De forma predeterminada, se recomienda el modo sin estado () stateless_http=True para los servidores MCP básicos. La plataforma agrega automáticamente un Mcp-Session-Id encabezado para cualquier solicitud que no lo contenga, de modo que los clientes de MCP puedan mantener la continuidad de la conexión con la misma sesión de Amazon Bedrock AgentCore Runtime.
En el caso de los servidores MCP que requieren interacciones en varios turnos (selección), LLM-generated contenido (muestreo) o notificaciones de progreso, el modo con estado () habilita estas funciones. stateless_http=False En el modo con estado, el motor de ejecución conserva el estado de la sesión MCP en todas las solicitudes de la misma invocación. Para obtener más información, consulte Características del servidor MCP de Stateful.
La carga útil de la InvokeAgentRuntime API se transfiere directamente, lo que permite transferir fácilmente los mensajes RPC de protocolos como el MCP.
Requisitos previos
Antes de implementar un servidor MCP, asegúrese de disponer de lo siguiente:
-
Python 3.10 o superior instalado y conocimientos básicos de Python
-
Una AWS cuenta con los permisos apropiados y las credenciales locales configuradas
Paso 1: Cree su servidor MCP
Instalación de los paquetes obligatorios
Primero, instale el paquete MCP:
pip install mcp
Cree su primer servidor MCP
Cree un archivo nuevo llamadomy_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")
Entendiendo el código
En el ejemplo se utilizan los siguientes componentes:
-
FastMCP: crea un servidor MCP que puede alojar sus herramientas
-
@mcp .tool (): decorador que convierte tus funciones de Python en herramientas de MCP
-
Herramientas: Tres herramientas sencillas que muestran diferentes tipos de operaciones
-
stateless_http=true: configura el servidor en modo sin estado, que es el predeterminado para los servidores MCP básicos
sugerencia
En el caso de los servidores MCP que requieren interacciones de varios turnos (selección) o contenido (muestreo), utilícelos para habilitar el modo con estado. LLM-generated stateless_http=False Los servidores MCP con estado mantienen el contexto de la sesión en varias solicitudes dentro de la misma herramienta de invocación. Para obtener más información, consulte Características del servidor MCP de Stateful.
Paso 2: Pruebe su servidor MCP localmente
Inicie su servidor MCP
Ejecute su servidor MCP localmente:
python my_mcp_server.py
Debería ver un resultado que indica que el servidor se está ejecutando en el puerto8000.
Realice una prueba con el cliente MCP
Desde una nueva terminal, cree un nuevo archivo my_mcp_client.py y ejecútelo 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())
También puede probar su servidor con el inspector MCP, tal como se describe en Pruebas locales con el inspector MCP.
Paso 3: Implemente su servidor MCP en AWS
Instale las herramientas de implementación
Instale la AgentCore CLI:
npm install -g @aws/agentcore
La AgentCore CLI se usa para implementar el agente en AgentCore Runtime.
Cree una carpeta de proyecto con la siguiente estructura:
## 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
Cree un nuevo archivo llamadorequirements.txt, añada lo siguiente:
mcp
requirements.txtespecifica los requisitos que el agente necesita para desplegarse en AgentCore Runtime.
Cree su proyecto para su implementación
Antes de crear el proyecto, debes configurar un grupo de usuarios de Cognito para la autenticación, tal y como se describe en Configurar el grupo de usuarios de Cognito para la autenticación. Esto proporciona los tokens de OAuth necesarios para un acceso seguro al servidor implementado.
nota
A partir del 7 de octubre de 2025, Amazon Bedrock AgentCore utilizará un Service-Linked rol para los permisos de identidad de las cargas de trabajo cuando utilice la autenticación OAuth. Para obtener información detallada sobre este cambio, consulte Función vinculada al servicio de identidad.
Tras configurar la autenticación, diseñe un nuevo proyecto con el protocolo 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 configuración de tiempo de ejecución CUSTOM_JWT y estructura la estructura del proyecto. Los comandos copian el servidor sobre el punto de app/MCPServer/main.py entrada generado y añaden su dependencia a. pyproject.toml
Implemente en AWS
Despliegue su agente:
agentcore deploy
Este comando hará lo siguiente:
-
Empaqueta tu código de agente y sus dependencias
-
Suba el artefacto de implementación a Amazon S3
-
Cree un entorno de ejecución de Amazon Bedrock AgentCore
-
Despliegue su agente para AWS
Tras la implementación, recibirá un ARN en tiempo de ejecución del agente con el siguiente aspecto:
arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_mcp_server-xyz123
Paso 4: Invoca tu servidor MCP implementado
Realice una prueba con el cliente MCP (remoto)
Antes de realizar la prueba, defina las siguientes variables de entorno:
-
Exporte el ARN del agente como variable de entorno:
export AGENT_ARN="agent_arn" -
Exporte el token del portador como variable de entorno:
export BEARER_TOKEN="bearer_token"
si pasas un Accept encabezado, este debe seguir el estándar MCP. application/json y. text/event-stream
Cree un nuevo archivo my_mcp_client_remote.py y ejecútelo 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())
También puede probar el servidor desplegado con el MCP Inspector, tal y como se describe en Pruebas remotas con MCP Inspector.
Respuestas de error de autenticación para los agentes OAuth-Configured
OAuth-configured los agentes siguen los estándares de autenticación RFC 6749 (OAuth 2.0).
401 No autorizado: falta la autenticación
Si no se proporciona ningún token de portador en el encabezado de autorización, la respuesta es:
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}"
Flujo de extremo a extremo con Auth0
En esta sección se muestra la autenticación de OAuth con Auth0 como proveedor de identidad. Usamos Auth0 en este ejemplo porque admite el registro dinámico de clientes (DCR), lo que simplifica el proceso de configuración del cliente al permitir que los clientes se registren por sí mismos mediante programación durante el tiempo de ejecución.
Paso 1 - Paso 3: Cree y pruebe su servidor MCP
Siga los pasos 1 a 3, desde el paso 1: crear su servidor MCP hasta el paso 3: implementar su servidor MCP AWS para crear y probar su servidor MCP.
Paso 4: Cree la aplicación Auth0
Siga las instrucciones de configuración de Auth0 en Auth0 de Okta.
Habilite el registro dinámico de clientes:
-
Panel de control → Configuración → Avanzado
-
Active la opción «Registro dinámico de aplicaciones en OIDC»
-
Grabación de los cambios
Para obtener más información, consulte la documentación sobre el registro dinámico de clientes de Auth0.
Paso 5: Crea tu proyecto para su implementación
Después de configurar la autenticación, diseñe un nuevo proyecto con el protocolo 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 ../..
Sustituya los marcadores de posición Auth0 por valores de su aplicación Auth0. La CLI crea una configuración de tiempo de ejecución CUSTOM_JWT y estructura la estructura del proyecto. Los comandos copian el servidor sobre el punto de app/MCPServer/main.py entrada generado y añaden su dependencia a. pyproject.toml
Paso 6: Despliegue en AWS
Despliegue su agente:
agentcore deploy
Este comando hará lo siguiente:
-
Empaqueta tu código de agente y sus dependencias
-
Suba el artefacto de implementación a Amazon S3
-
Cree un tiempo de ejecución de Amazon Bedrock AgentCore
-
Despliegue su agente para AWS
Tras la implementación, recibirá un ARN en tiempo de ejecución del agente con el siguiente aspecto:
arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_mcp_server-xyz123
Paso 7: Invoca al agente desplegado
Este cliente se basa en el ejemplo oficial de cliente simple de autenticación simple del MCP SDK, con modificaciones.
nota
Cuando utilices Auth0 con el registro dinámico de clientes, debes incluir el audience parámetro en las solicitudes de autorización para recibir los tokens de JWT. Sin este parámetro, Auth0 devuelve los tokens opacos o los tokens JWE (cifrados) en lugar de los tokens JWT estándar. El SDK de MCP envía el parámetro de OAuth 2.0 (RFC 8707), pero Auth0 requiere el resource parámetro OIDC para los tokens de JWT. audience Ambos parámetros tienen propósitos audience similares, pero Auth0 prioriza. Para obtener más información, consulte Auth0 Community: tokens JWT con registro dinámico de aplicaciones.
Cree un archivo llamado mcp_auth0_client.py con el siguiente código. Este cliente gestiona Auth0-specific los requisitos, incluido el parámetro de audiencia:
nota
El código incluye la aplicación de parches httpx para inyectar User-Agent encabezados en todas las solicitudes HTTP. Esto es necesario porque el SDK de Python de MCP actualmente no incluye User-Agent encabezados en sus solicitudes HTTP, lo que puede provocar problemas con AWS las reglas del WAF que requieren encabezados. User-Agent Para obtener más información, consulte el número #1664 del SDK de Python de MCP y los grupos de reglas administrados por WAF.
#!/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()
Para usar el cliente:
-
Establezca las variables de entorno necesarias:
export AGENT_ARN="arn:aws:bedrock:us-west-2:123456789012:agent/ABCD1234" -
Establezca la variable de Auth0-specific entorno (requerida solo para Auth0):
export AUTH0_API_IDENTIFIER="your-api-identifier" -
Ejecute el cliente:
python mcp_auth0_client.py
El cliente realizará automáticamente lo siguiente:
-
Codifique el ARN del agente para usarlo en la URL
-
Cree la URL del punto final de invocación del MCP
-
Agregue el
audienceparámetro Auth0 a las solicitudes de autorización (cuando utilice Auth0) -
Trabaja con cualquier proveedor de identidad compatible con OAuth 2.0
Apéndice
Configure el grupo de usuarios de Cognito para la autenticación
Cree un archivo nuevo setup_cognito.sh y añada el siguiente contenido.
#!/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"
Abra una ventana de terminal y defina las siguientes variables de entorno:
-
REGION— la AWS región que desea usar -
USERNAME— el nombre de usuario del nuevo usuario -
PASSWORD— la contraseña del nuevo usuario
export REGION=us-east-1 # Set your desired Region export USERNAME="user-name" export PASSWORD="password"
Ejecute el script con el comandosource setup_cognito.sh.
nota
Para obtener información detallada sobre la configuración de la autenticación OAuth y las Service-Linked funciones, consulte Autenticar y autorizar con la autenticación entrante y la autenticación saliente.
Tras ejecutar este script, anote los siguientes valores para usarlos en la configuración de implementación:
-
URL de descubrimiento: se utiliza durante el
agentcore createpaso -
ID de cliente: se usó durante el
agentcore createpaso -
Token de portador: se usa al invocar el servidor desplegado
Pruebas locales con el inspector MCP
El MCP Inspector es una herramienta visual para probar los servidores MCP. Para usarlo, necesita:
-
Node.js y npm instalado
Instale y ejecute el MCP Inspector:
npx @modelcontextprotocol/inspector
Esto hará lo siguiente:
-
Inicie el servidor MCP Inspector
-
Muestra una URL en tu terminal (normalmente
http://localhost:6274)
Para usar el Inspector:
-
Navega hasta
http://localhost:6274en tu navegador -
Pegue la URL del servidor MCP (
http://localhost:8000/mcp) en el campo de conexión del Inspector MCP -
Verás tus herramientas en la barra lateral
-
Haz clic en cualquier herramienta para probarla
-
Rellene los parámetros (p. ej., para
add_numbers, introduzca valores paraayb) -
Haga clic en «Call Tool» para ver el resultado
Pruebas remotas con el inspector MCP
También puede probar el servidor implementado con el MCP Inspector. En primer lugar, el URL-encode ARN de su agente:
export AGENT_ARN="arn:aws:bedrock-agentcore:us-west-2:123456789012:runtime/my_mcp_server-xyz123" echo -n $AGENT_ARN | jq -sRr '@uri'
Esto genera el URL-encoded ARN:
arn%3Aaws%3Abedrock-agentcore%3Aus-west-2%3A123456789012%3Aruntime%2Fmy_mcp_server-xyz123
A continuación, conéctese con el MCP Inspector:
-
Inicie el MCP Inspector:
npx @modelcontextprotocol/inspector -
En la interfaz web:
-
Seleccione «HTTP transmisible» como transporte
-
Introduzca la URL del punto final de su agente con el ARN codificado. Asegúrese de usar la misma región que el ARN de su agente:
https://bedrock-agentcore.REGION.amazonaws.com/runtimes/ENCODED_ARN/invocations?qualifier=DEFAULTEjemplo para 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 -
Añade tu token de portador en la sección de autenticación con el nombre y el valor del encabezado
AuthorizationBearer YOUR_TOKEN -
Haz clic en «Conectar»
-
-
Pon a prueba tus herramientas igual que lo hiciste localmente
