# 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](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.](runtime-mcp-protocol-contract.md)
**Topics**
+ [Cómo AgentCore apoya Amazon Bedrock a MCP](#runtime-mcp-how-it-works)
+ [Requisitos previos](#runtime-mcp-prerequisites)
+ [Paso 1: Cree su servidor MCP](#runtime-mcp-create-server)
+ [Paso 2: Pruebe su servidor MCP localmente](#runtime-mcp-test-locally)
+ [Paso 3: Implemente su servidor MCP en AWS](#runtime-mcp-deploy-aws)
+ [Paso 4: invoque el servidor MCP desplegado](#runtime-mcp-invoke-server)
+ [Respuestas de error de autenticación para los agentes OAuth-Configured](#runtime-mcp-auth-error-responses)
+ [Flujo de extremo a extremo con Auth0](#runtime-mcp-auth0-flow)
+ [Apéndice](#runtime-mcp-appendix)
## 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 ruta`0.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.](mcp-stateful-features.md)
La carga útil de la [InvokeAgentRuntime](https://docs.aws.amazon.com/bedrock-agentcore/latest/APIReference/API_InvokeAgentRuntime.html)API 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 llamado`my_mcp_server.py`:
```
# my_mcp_server.py
from mcp.server.fastmcp import FastMCP
from starlette.responses import JSONResponse
mcp = FastMCP(host="0.0.0.0", stateless_http=True)
@mcp.tool()
def add_numbers(a: int, b: int) -> int:
"""Add two numbers together"""
return a + b
@mcp.tool()
def multiply_numbers(a: int, b: int) -> int:
"""Multiply two numbers together"""
return a * b
@mcp.tool()
def greet_user(name: str) -> str:
"""Greet a user by name"""
return f"Hello, {name}! Nice to meet you."
if __name__ == "__main__":
mcp.run(transport="streamable-http")
```
### 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.](mcp-stateful-features.md)
## 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 puerto`8000`.
### 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](#runtime-mcp-appendix-b).
## 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.txt`especifica 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](#runtime-mcp-appendix-a) 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](service-linked-roles.md#identity-service-linked-role).
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:
1. Package su código de agente y sus dependencias
1. Cargue el artefacto de implementación en Amazon S3
1. Cree un entorno de ejecución de Amazon Bedrock AgentCore
1. 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.](https://modelcontextprotocol.io/specification/2025-06-18/basic/transports#sending-messages-to-the-server) Los tipos de medios aceptables son `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](#runtime-mcp-appendix-c).
## 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](https://datatracker.ietf.org/doc/html/rfc6749)). Cuando falta la autenticación, el servicio devuelve una respuesta 401 no autorizada con un WWW-Authenticate encabezado (según la [RFC 7235](https://datatracker.ietf.org/doc/html/rfc7235)), lo que permite a los clientes descubrir los puntos finales del servidor de autorización a través de la API. GetRuntimeProtectedResourceMetadata
### 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](#runtime-mcp-create-server) hasta el [paso 3: Implemente su servidor MCP AWS para](#runtime-mcp-deploy-aws) crear y probar su servidor MCP.
### Paso 4: Cree la aplicación Auth0
Siga las instrucciones de configuración de Auth0 en [Auth0](identity-idp-auth0.md) de Okta.
**Habilite el registro dinámico de clientes:**
1. Panel de control → Configuración → Avanzado
1. Active la opción «Registro dinámico de aplicaciones OIDC» →
1. Grabación de los cambios
Para obtener más información, consulte la documentación de registro [dinámico de clientes de Auth0](https://auth0.com/docs/get-started/applications/dynamic-client-registration).
### 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](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/clients/simple-auth-client/mcp_simple_auth_client/main.py). Auth0-specific
**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](https://community.auth0.com/t/jwt-tokens-with-dynamic-application-registration/189741) 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 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](https://github.com/modelcontextprotocol/python-sdk/issues/1664) y los grupos de [reglas gestionados por AWS WAF](https://docs.aws.amazon.com/waf/latest/developerguide/aws-managed-rule-groups-baseline.html).
#### Código de cliente MCP Python Auth0
```
#!/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"""
Authorization Successful!
You can close this window and return to the terminal.
""")
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"""
Authorization Failed
Error: {query_params["error"][0]}
You can close this window and return to the terminal.
""".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 [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 ', 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:
1. Establezca las variables de entorno necesarias:
```
export AGENT_ARN="arn:aws:bedrock:us-west-2:123456789012:agent/ABCD1234"
```
1. Establezca la variable de Auth0-specific entorno (necesaria solo para Auth0):
```
export AUTH0_API_IDENTIFIER="your-api-identifier"
```
1. 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 `audience` pará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 comando`source 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](runtime-oauth.md).
Tras ejecutar este script, anote los siguientes valores para utilizarlos en la configuración de despliegue:
+ URL de descubrimiento: se utilizó durante el `agentcore create` paso
+ ID de cliente: se utilizó durante el `agentcore create` paso
+ 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:
1. Navegue hasta `http://localhost:6274` en su navegador
1. Pegue la URL del servidor MCP (`http://localhost:8000/mcp`) en el campo de conexión del Inspector MCP
1. Verás tus herramientas listadas en la barra lateral
1. Haz clic en cualquier herramienta para probarla
1. Rellene los parámetros (p. ej., para`add_numbers`, introduzca los valores para `a` y`b`)
1. 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:
1. Inicie el Inspector MCP:
```
npx @modelcontextprotocol/inspector
```
1. 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=DEFAULT
```
Ejemplo 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 `Authorization` `Bearer YOUR_TOKEN`
+ Haga clic en «Conectar»
1. Pruebe sus herramientas tal como lo hizo localmente