As traduções são geradas por tradução automática. Em caso de conflito entre o conteúdo da tradução e da versão original em inglês, a versão em inglês prevalecerá.
Implemente servidores MCP em AgentCore tempo de execução
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 obter um exemplo, consulte Fundamentos do servidor AgentCore MCP em. GitHub
Nesta seção, você aprende:
-
Como criar um servidor MCP com ferramentas
-
Como testar seu servidor localmente
-
Como implantar seu servidor no AWS
-
Como invocar seu servidor implantado
Para obter mais informações sobre o MCP, consulte o 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 streamáveis sem estado e com estado. Por padrão, o modo sem estado (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 do 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 de 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 InvokeAgentRuntime API é transmitida diretamente, permitindo que mensagens RPC de protocolos como o MCP sejam facilmente transmitidas por proxy.
Pré-requisitos
Antes de implantar um servidor MCP, verifique se você tem o seguinte:
-
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
O exemplo usa os seguintes componentes:
-
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árias voltas (elicitação) ou LLM-generated conteúdo (amostragem), use stateless_http=False para ativar o modo de 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 uma 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: Implante seu servidor MCP em AWS
Instale ferramentas de implantação
Instale a AgentCore CLI:
npm install -g @aws/agentcore
Você usa a AgentCore CLI para implantar seu agente no AgentCore Runtime.
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 chamado requirements.txt e 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 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 --project-name MCPServerProject --no-agent cd MCPServerProject agentcore add agent \ --name MCPServer \ --language Python \ --protocol MCP \ --authorizer-type CUSTOM_JWT \ --discovery-url "https://cognito-idp.$REGION.amazonaws.com/$POOL_ID/.well-known/openid-configuration" \ --allowed-clients "$CLIENT_ID" \ --request-header-allowlist Authorization cp ../my_mcp_server.py app/MCPServer/main.py cd app/MCPServer uv add mcp cd ../..
A CLI cria uma configuração de tempo de execução CUSTOM_JWT e estrutura a estrutura do projeto. Os comandos copiam seu servidor sobre o app/MCPServer/main.py ponto de entrada gerado e adicionam sua dependência a. pyproject.toml
Implantar em AWS
Implante seu agente:
agentcore deploy
Esse comando irá:
-
Empacote 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 ambiente 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 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), 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: Crie e teste seu servidor MCP
Siga as etapas 1 a 3 da Etapa 1: Crie seu servidor MCP até a Etapa 3: Implante 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 by Okta.
Ativar 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 de registro dinâmico de clientes do
Etapa 5: crie seu projeto para implantação
Depois de configurar a autenticação, crie um novo projeto com o protocolo MCP:
agentcore create --project-name MCPServerProject --no-agent cd MCPServerProject agentcore add agent \ --name MCPServer \ --language Python \ --protocol MCP \ --authorizer-type CUSTOM_JWT \ --discovery-url "<AUTH0_DISCOVERY_URL>" \ --allowed-clients "<AUTH0_CLIENT_ID>" \ --request-header-allowlist Authorization cp ../my_mcp_server.py app/MCPServer/main.py cd app/MCPServer uv add mcp cd ../..
Substitua os espaços reservados do Auth0 pelos valores do seu aplicativo Auth0. A CLI cria uma configuração de tempo de execução CUSTOM_JWT e estrutura a estrutura do projeto. Os comandos copiam seu servidor sobre o app/MCPServer/main.py ponto de entrada gerado e adicionam sua dependência a. pyproject.toml
Etapa 6: implantar em AWS
Implante seu agente:
agentcore deploy
Esse comando irá:
-
Empacote 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 ambiente 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 com modificações.
nota
Ao usar o Auth0 com o registro dinâmico de clientes, você deve incluir o audience parâmetro nas solicitações de autorização para receber tokens JWT. Sem esse parâmetro, o Auth0 retorna tokens opacos ou tokens JWE (criptografados) em vez dos tokens JWT padrão. O SDK MCP 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 os 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 inclui User-Agent cabeçalhos em suas solicitações HTTP, o que pode causar problemas com as regras do AWS WAF que exigem cabeçalhos. User-Agent Para obter mais informações, consulte a edição #1664 do SDK MCP Python
#!/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 automaticamente:
-
Codifique o ARN do agente para uso na URL
-
Construa o URL do endpoint de invocação do MCP
-
Adicionar 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 para o 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: usada 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 MCP Inspector:
npx @modelcontextprotocol/inspector
Isso vai:
-
Inicie o servidor MCP Inspector
-
Exibir um URL em seu terminal (normalmente
http://localhost:6274)
Para usar o Inspetor:
-
Navegue até
http://localhost:6274em seu navegador -
Cole o URL do servidor MCP (
http://localhost:8000/mcp) no campo de conexão do MCP Inspector -
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 MCP Inspector:
-
Inicie o MCP Inspector:
npx @modelcontextprotocol/inspector -
Na interface da web:
-
Selecione “Streamable HTTP” 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 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 -
Adicione seu token de portador na seção Autenticação com nome
Authorizatione valor do cabeçalhoBearer YOUR_TOKEN -
Clique em “Conectar”
-
-
Teste suas ferramentas da mesma forma que você fez localmente
