Implemente servidores MCP no AgentCore Runtime
O Amazon Bedrock AgentCore Runtime permite que você implante e execute servidores Model Context Protocol (MCP) no AgentCore Runtime. Este guia explica como criar, testar e implantar seu primeiro servidor MCP.
Para ver um exemplo, consulte https://github.com/awslabs/amazon-bedrock-agentcore-samples/tree/main/01-tutorials/01-AgentCore-runtime/02-hosting-MCP-server
Nesta seção, você aprende:
-
Como criar um servidor MCP com ferramentas
-
Como testar seu servidor localmente
-
Como implantar seu servidor em AWS
-
Como invocar seu servidor implantado
Para obter mais informações sobre MCP, consulte Contrato de protocolo MCP.
Como o Amazon Bedrock AgentCore oferece suporte ao MCP
Quando você configura um Amazon Bedrock AgentCore Runtime com o protocolo MCP, o serviço espera que os contêineres do servidor MCP estejam disponíveis no caminho0.0.0.0:8000/mcp, que é o caminho padrão suportado pela maioria dos SDKs oficiais do servidor MCP.
O Amazon Bedrock AgentCore oferece suporte a servidores MCP HTTP transmissíveis e sem estado. Por padrão, o modo stateless (stateless_http=True) é recomendado para servidores MCP básicos. A plataforma adiciona automaticamente um Mcp-Session-Id cabeçalho para qualquer solicitação sem um, para que os clientes MCP possam manter a continuidade da conexão com a mesma sessão do Amazon Bedrock AgentCore Runtime.
Para servidores MCP que exigem interações em vários turnos (elicitação), LLM-generated conteúdo (amostragem) ou notificações de progresso, o modo stateful () ativa esses recursos. stateless_http=False No modo com estado, o tempo de execução preserva o estado da sessão MCP em todas as solicitações dentro da mesma invocação. Para obter mais informações, consulte Recursos do servidor Stateful MCP.
A carga útil da InvokeAgentRuntimeAPI é transmitida diretamente, permitindo que mensagens RPC de protocolos como MCP sejam facilmente enviadas por proxy.
Pré-requisitos
-
Python 3.10 ou superior instalado e compreensão básica do Python
-
Uma AWS conta com permissões apropriadas e credenciais locais configuradas
Etapa 1: Crie seu servidor MCP
Instalar os pacotes obrigatórios
Primeiro, instale o pacote MCP:
pip install mcp
Crie seu primeiro servidor MCP
Crie um novo arquivo chamadomy_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")
Entendendo o código
-
FastMCP: cria um servidor MCP que pode hospedar suas ferramentas
-
@mcp .tool (): Decorador que transforma suas funções Python em ferramentas MCP
-
Ferramentas: três ferramentas simples que demonstram diferentes tipos de operações
-
stateless_http=True: configura o servidor no modo sem estado, que é o padrão para servidores MCP básicos
dica
Para servidores MCP que exigem interações de vários turnos (elicitação) ou LLM-generated conteúdo (amostragem), use stateless_http=False para ativar o modo com estado. Os servidores MCP com estado mantêm o contexto da sessão em várias solicitações na mesma invocação de ferramenta. Para obter mais informações, consulte Recursos do servidor Stateful MCP.
Etapa 2: Teste seu servidor MCP localmente
Inicie seu servidor MCP
Execute seu servidor MCP localmente:
python my_mcp_server.py
Você deve ver a saída indicando que o servidor está sendo executado na porta8000.
Teste com o cliente MCP
Em um novo terminal, crie um novo arquivo my_mcp_client.py e execute-o 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())
Você também pode testar seu servidor usando o Inspetor MCP, conforme descrito em Teste local com o inspetor MCP.
Etapa 3: implantar seu servidor MCP em AWS
Instalar ferramentas de implantação
Instale a AgentCore CLI:
npm install -g @aws/agentcore
Você usa a AgentCore CLI para implantar seu agente no Runtime. AgentCore
Crie uma pasta de projeto com a seguinte estrutura:
## 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
Crie um novo arquivo chamadorequirements.txt, adicione o seguinte a ele:
mcp
requirements.txtespecifica os requisitos que o agente precisa para implantação no AgentCore Runtime.
Crie seu projeto para implantação
Antes de criar seu projeto, você precisa configurar um grupo de usuários do Cognito para autenticação, conforme descrito em Configurar o grupo de usuários do Cognito para autenticação. Isso fornece os tokens OAuth necessários para acesso seguro ao seu servidor implantado.
nota
A partir de 7 de outubro de 2025, o Amazon Bedrock AgentCore usa uma Service-Linked função para permissões de identidade de carga de trabalho ao usar a autenticação OAuth. Para obter informações detalhadas sobre essa alteração, consulte Função vinculada ao serviço de identidade.
Depois de configurar a autenticação, crie um novo projeto com o protocolo MCP:
agentcore create --protocol MCP
Siga as instruções interativas para fornecer o nome do projeto. A CLI estrutura a estrutura do projeto, incluindo um arquivo de configuração. agentcore/agentcore.json Copie seu my_mcp_server.py arquivo no diretório de código do agente do projeto gerado e garanta que o ponto de entrada em agentcore/agentcore.json pontos para o arquivo do servidor.
Implemente em AWS
Implante seu agente:
agentcore deploy
Esse comando irá:
-
Package o código e as dependências do seu agente
-
Faça o upload do artefato de implantação para o Amazon S3
-
Crie um tempo de execução do Amazon Bedrock AgentCore
-
Implante seu agente para AWS
Após a implantação, você receberá um ARN de tempo de execução do agente que se parece com:
arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_mcp_server-xyz123
Etapa 4: Invocar seu servidor MCP implantado
Teste com o cliente MCP (remoto)
Antes de testar, defina as seguintes variáveis de ambiente:
-
Exporte o ARN do agente como uma variável de ambiente:
export AGENT_ARN="agent_arn" -
Exporte o token do portador como uma variável de ambiente:
export BEARER_TOKEN="bearer_token"
se você passar um Accept cabeçalho, ele deverá seguir o padrão MCP.application/json text/event-stream e.
Crie um novo arquivo my_mcp_client_remote.py e execute-o 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())
Você também pode testar seu servidor implantado usando o Inspetor MCP, conforme descrito em Teste remoto com o inspetor MCP.
Respostas de erro de autenticação para OAuth-Configured agentes
OAuth-configured os agentes seguem os padrões de autenticação RFC 6749 (OAuth 2.0
401 Não autorizado - Autenticação ausente
Quando nenhum token de portador é fornecido no cabeçalho de autorização, a resposta é:
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}"
Fluxo de ponta a ponta com Auth0
Esta seção demonstra a autenticação OAuth usando Auth0 como provedor de identidade. Usamos o Auth0 neste exemplo porque ele oferece suporte ao Registro Dinâmico de Clientes (DCR), o que simplifica o processo de configuração do cliente ao permitir que os clientes se registrem programaticamente em tempo de execução.
Etapa 1 - Etapa 3: criar e testar seu servidor MCP
Siga as etapas 1 a 3 da Etapa 1: Criar seu servidor MCP até a Etapa 3: Implantar seu servidor MCP AWS para criar e testar seu servidor MCP.
Etapa 4: criar o aplicativo Auth0
Siga as instruções de configuração do Auth0 em Auth0 da Okta.
Ativar o registro dinâmico de clientes:
-
Painel → Configurações → Avançado
-
Alterne “Registro dinâmico de aplicativos OIDC” → ATIVADO
-
Salve as alterações.
Para obter mais informações, consulte a documentação do Auth0 Dynamic Client Registration
Etapa 5: criar seu projeto para implantação
Depois de configurar a autenticação, crie um novo projeto com o protocolo MCP:
agentcore create --protocol MCP
Siga as instruções interativas para fornecer o nome do projeto. A CLI estrutura a estrutura do projeto, incluindo um arquivo de configuração. agentcore/agentcore.json Copie seu my_mcp_server.py arquivo no diretório de código do agente do projeto gerado e garanta que o ponto de entrada em agentcore/agentcore.json pontos para o arquivo do servidor.
Etapa 6: implantar em AWS
Implante seu agente:
agentcore deploy
Esse comando irá:
-
Package o código e as dependências do seu agente
-
Faça o upload do artefato de implantação para o Amazon S3
-
Crie um tempo de execução do Amazon Bedrock AgentCore
-
Implante seu agente para AWS
Após a implantação, você receberá um ARN de tempo de execução do agente que se parece com:
arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_mcp_server-xyz123
Etapa 7: invocar seu agente implantado
Esse cliente é baseado no exemplo oficial do MCP SDK simple-auth-client
nota
Ao usar Auth0 com o Registro Dinâmico de Cliente, você deve incluir o audience parâmetro nas solicitações de autorização para receber tokens JWT. Sem esse parâmetro, Auth0 retorna tokens opacos ou tokens JWE (criptografados) em vez de tokens JWT padrão. O MCP SDK envia o parâmetro do OAuth 2.0 (RFC 8707), mas o Auth0 exige o resource parâmetro OIDC para tokens JWT. audience Ambos os parâmetros têm propósitos semelhantes, mas o Auth0 audience prioriza. Para obter mais informações, consulte Comunidade Auth0 - Tokens JWT com registro dinâmico de aplicativos
Crie um arquivo denominado mcp_auth0_client.py com o código a seguir. Esse cliente lida com Auth0-specific requisitos, incluindo o parâmetro de audiência:
nota
O código inclui patches httpx para injetar User-Agent cabeçalhos em todas as solicitações HTTP. Isso é necessário porque o SDK MCP Python atualmente não User-Agent inclui cabeçalhos em suas solicitações HTTP, o que pode causar problemas AWS com as regras do WAF que exigem cabeçalhos. User-Agent Para obter mais informações, consulte o problema #1664 do SDK do MCP Python AWS
#!/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 o cliente:
-
Defina as variáveis de ambiente necessárias:
export AGENT_ARN="arn:aws:bedrock:us-west-2:123456789012:agent/ABCD1234" -
Defina a variável de Auth0-specific ambiente (necessária somente para Auth0):
export AUTH0_API_IDENTIFIER="your-api-identifier" -
Execute o cliente:
python mcp_auth0_client.py
O cliente irá automaticamente:
-
Codifique o ARN do agente para uso na URL
-
Construa a URL do endpoint de invocação do MCP
-
Adicione o
audienceparâmetro Auth0 às solicitações de autorização (ao usar Auth0) -
Trabalhe com qualquer provedor de identidade compatível com OAuth 2.0
Apêndice
Configurar o grupo de usuários do Cognito para autenticação
Crie um novo arquivo setup_cognito.sh e adicione o conteúdo a seguir.
#!/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 uma janela de terminal e defina as seguintes variáveis de ambiente:
-
REGION— a AWS região que você deseja usar -
USERNAME— o nome de usuário do novo usuário -
PASSWORD— a senha do novo usuário
export REGION=us-east-1 // set your desired Region export USERNAME=USER NAME export PASSWORD=PASSWORD
Execute o script usando o comandosource setup_cognito.sh.
nota
Para obter informações detalhadas sobre a configuração da autenticação OAuth e a Service-Linked função, consulte Autenticar e autorizar com Autenticação de Entrada e Autenticação de Saída.
Depois de executar esse script, observe os seguintes valores para uso na configuração de implantação:
-
URL de descoberta: usado durante a
agentcore createetapa -
ID do cliente: usado durante a
agentcore createetapa -
Token do portador: usado ao invocar seu servidor implantado
Teste local com o inspetor MCP
O MCP Inspector é uma ferramenta visual para testar servidores MCP. Para usá-lo, você precisa:
-
Node.js e npm instalado
Instale e execute o Inspector MCP:
npx @modelcontextprotocol/inspector
Isso vai:
-
Inicie o servidor MCP Inspector
-
Exibir uma URL em seu terminal (normalmente
http://localhost:6274)
Para usar o Inspector:
-
Navegue até
http://localhost:6274no seu navegador -
Cole o URL do servidor MCP (
http://localhost:8000/mcp) no campo de conexão do Inspector MCP -
Você verá suas ferramentas listadas na barra lateral
-
Clique em qualquer ferramenta para testá-la
-
Preencha os parâmetros (por exemplo, para
add_numbers, insira valores paraaeb) -
Clique em “Ferramenta de chamada” para ver o resultado
Teste remoto com o inspetor MCP
Você também pode testar seu servidor implantado usando o MCP Inspector. Primeiro, URL-encode seu agente ARN:
export AGENT_ARN="arn:aws:bedrock-agentcore:us-west-2:123456789012:runtime/my_mcp_server-xyz123" echo -n $AGENT_ARN | jq -sRr '@uri'
Isso gera o URL-encoded ARN:
arn%3Aaws%3Abedrock-agentcore%3Aus-west-2%3A123456789012%3Aruntime%2Fmy_mcp_server-xyz123
Em seguida, conecte-se com o Inspetor MCP:
-
Inicie o Inspector MCP:
npx @modelcontextprotocol/inspector -
Na interface da web:
-
Selecione “HTTP Streamable” como transporte
-
Insira o URL do endpoint do seu agente usando o ARN codificado. Certifique-se de usar a mesma região do ARN do seu agente:
https://bedrock-agentcore.REGION.amazonaws.com/runtimes/ENCODED_ARN/invocations?qualifier=DEFAULTExemplo de 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 -
Adicione seu token de portador na seção Autenticação com nome
Authorizatione valor do cabeçalhoBearer YOUR_TOKEN -
Clique em “Connect”
-
-
Teste suas ferramentas da mesma forma que você fez localmente
