Implemente servidores MCP en tiempo de ejecución AgentCore
Amazon Bedrock AgentCore Runtime le permite implementar y ejecutar servidores del Model Context Protocol (MCP) en el AgentCore entorno de ejecución. Esta guía explica cómo crear, probar e implementar su primer servidor MCP.
Para ver un ejemplo, consulta https://github.com/awslabs/amazon-bedrock-agentcore-samples/tree/main/01-tutorials/01-AgentCore-runtime/02-hosting-MCP-server
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 el MCP, consulte el contrato de protocolo MCP.
Cómo AgentCore apoya Amazon Bedrock 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 de servidores MCP oficiales.
Amazon Bedrock AgentCore es compatible con servidores MCP HTTP con capacidad de estado y sin estado. De forma predeterminada, se recomienda el modo sin estado () stateless_http=True para los servidores MCP básicos. La plataforma añade automáticamente un Mcp-Session-Id encabezado para cualquier solicitud que no lo tenga, de forma que los clientes 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 múltiples (obtención de información), notificaciones de LLM-generated contenido (muestreo) o de progreso, el modo con estado () habilita estas funciones. stateless_http=False En el modo con estado, el tiempo de ejecución conserva el estado de la sesión del MCP en todas las solicitudes de la misma invocación. Para obtener más información, consulte Características del servidor MCP con estado.
La carga útil de la InvokeAgentRuntimeAPI se transfiere directamente, lo que permite procesar fácilmente los mensajes RPC de protocolos como el MCP.
Requisitos previos
-
Python 3.10 o superior instalado y conocimientos básicos de Python
-
Una AWS cuenta con los permisos adecuados 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")
Entender el código
-
FastMCP: crea un servidor MCP que puede alojar sus herramientas
-
@mcp .tool (): Decorador que convierte tus funciones de Python en herramientas 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 (elicitació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 invocación de la herramienta. Para obtener más información, consulte Características del servidor MCP con estado.
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 la 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 y como se describe en Pruebas locales con el inspector MCP.
Paso 3: Implemente su servidor MCP en AWS
Instale las herramientas de despliegue
Instale la AgentCore CLI:
npm install -g @aws/agentcore
La AgentCore CLI se utiliza 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 llamado requirements.txt y agréguele lo siguiente:
mcp
requirements.txtespecifica los requisitos que el agente necesita para su despliegue en AgentCore Runtime.
Cree su proyecto para su despliegue
Antes de crear el proyecto, debe configurar un grupo de usuarios de Cognito para la autenticación, tal 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 la carga 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 --protocol MCP
Siga las instrucciones interactivas para proporcionar un nombre de proyecto. La CLI estructura el proyecto e incluye un archivo de agentcore/agentcore.json configuración. Copie el my_mcp_server.py archivo en el directorio de códigos de agente del proyecto generado y asegúrese de que el punto de entrada apunte al archivo del agentcore/agentcore.json servidor.
Implemente en AWS
Despliegue a su agente:
agentcore deploy
Este comando hará lo siguiente:
-
Package su código de agente y sus dependencias
-
Cargue el artefacto de implementación en Amazon S3
-
Cree un entorno de ejecución de Amazon Bedrock AgentCore
-
Despliegue a su agente en AWS
Tras la implementación, recibirá un ARN de 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: invoque el servidor MCP desplegado
Realice la prueba con el cliente MCP (remoto)
Antes de realizar la prueba, defina las siguientes variables de entorno:
-
El ARN del agente de exportación como variable de entorno:
export AGENT_ARN="agent_arn" -
Exporte el token portador como variable de entorno:
export BEARER_TOKEN="bearer_token"
si pasa un Accept encabezado, debe seguir el estándar MCP.application/json y. text/event-stream
Cree un archivo nuevo 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 mediante el Inspector de MCP, tal y como se describe en Pruebas remotas con el inspector de MCP.
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
Cuando no se proporciona ningún identificador 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 para este ejemplo porque admite el registro dinámico de clientes (DCR), que simplifica el proceso de configuración del cliente al permitir que los clientes se registren ellos mismos mediante programación en tiempo de ejecución.
Paso 1 - Paso 3: Cree y pruebe su servidor MCP
Siga los pasos 1 a 3, desde el paso 1: Cree su servidor MCP hasta el paso 3: Implemente 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 OIDC» →
-
Grabación de los cambios
Para obtener más información, consulte la documentación de registro dinámico de clientes de Auth0
Paso 5: Cree su proyecto para su implementación
Después de configurar la autenticación, diseñe un nuevo proyecto con el protocolo MCP:
agentcore create --protocol MCP
Siga las instrucciones interactivas para proporcionar un nombre de proyecto. La CLI estructura el proyecto e incluye un archivo de agentcore/agentcore.json configuración. Copie el my_mcp_server.py archivo en el directorio de códigos de agente del proyecto generado y asegúrese de que el punto de entrada apunte al archivo del agentcore/agentcore.json servidor.
Paso 6: Implementar en AWS
Despliegue a su agente:
agentcore deploy
Este comando hará lo siguiente:
-
Package su código de agente y sus dependencias
-
Cargue el artefacto de implementación en Amazon S3
-
Cree un entorno de ejecución de Amazon Bedrock AgentCore
-
Despliegue a su agente en AWS
Tras la implementación, recibirá un ARN de 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 el agente desplegado
Este cliente se basa en el ejemplo oficial de cliente de autenticación simple del SDK de MCP con modificaciones
nota
Si utiliza Auth0 con el registro dinámico de clientes, debe 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 JWT. audience Ambos parámetros sirven para fines similares, pero Auth0 prioriza. audience Para obtener más información, consulte Comunidad Auth0: Tokens JWT
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 un parche httpx para inyectar User-Agent encabezados en todas las solicitudes HTTP. Esto es necesario porque el SDK para Python de MCP actualmente no incluye User-Agent encabezados en sus solicitudes HTTP, lo que puede provocar problemas con las reglas de AWS WAF que requieren encabezados. User-Agent Para obtener más información, consulte el número #1664 del SDK de Python de 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()
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 (necesaria solo para Auth0):
export AUTH0_API_IDENTIFIER="your-api-identifier" -
Ejecute el cliente:
python mcp_auth0_client.py
El cliente hará lo siguiente automáticamente:
-
Codifique el ARN del agente para usarlo en la URL
-
Construya la URL del punto final de invocación del MCP
-
Agregue el
audienceparámetro Auth0 a las solicitudes de autorización (cuando utilice Auth0) -
Trabaje con cualquier proveedor de identidad compatible con OAuth 2.0
Apéndice
Configurar 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 configure 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 mediante el comandosource setup_cognito.sh.
nota
Para obtener información detallada sobre la configuración de la autenticación de OAuth y la Service-Linked función, consulte Autenticar y autorizar con autenticación entrante y autenticación saliente.
Tras ejecutar este script, anote los siguientes valores para utilizarlos en la configuración de despliegue:
-
URL de descubrimiento: se utilizó durante el
agentcore createpaso -
ID de cliente: se utilizó durante el
agentcore createpaso -
Token de portador: se usa al invocar el servidor desplegado
Pruebas locales con el inspector MCP
El Inspector MCP es una herramienta visual para probar servidores MCP. Para usarlo, necesitas:
-
Node.js y npm instalado
Instale y ejecute el Inspector MCP:
npx @modelcontextprotocol/inspector
Esto permitirá:
-
Inicie el servidor del Inspector MCP
-
Muestra una URL en tu terminal (normalmente
http://localhost:6274)
Para usar el Inspector:
-
Navegue hasta
http://localhost:6274en su navegador -
Pegue la URL del servidor MCP (
http://localhost:8000/mcp) en el campo de conexión del Inspector MCP -
Verás tus herramientas listadas en la barra lateral
-
Haz clic en cualquier herramienta para probarla
-
Rellene los parámetros (p. ej., para
add_numbers, introduzca los valores paraayb) -
Haga clic en «Herramienta de llamadas» para ver el resultado
Pruebas remotas con el inspector MCP
También puede probar el servidor desplegado mediante el Inspector MCP. En primer lugar, el ARN de URL-encode 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 Inspector de MCP:
-
Inicie el Inspector MCP:
npx @modelcontextprotocol/inspector -
En la interfaz web:
-
Seleccione «Streamable HTTP» como transporte
-
Introduzca la URL del punto final de su agente mediante 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 -
Agrega tu token de portador en la sección de autenticación con el nombre y el valor del encabezado
AuthorizationBearer YOUR_TOKEN -
Haga clic en «Conectar»
-
-
Pruebe sus herramientas tal como lo hizo localmente
