# Implemente servidores MCP no AgentCore Runtime
<a name="runtime-mcp"></a>

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](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.](runtime-mcp-protocol-contract.md)

**Topics**
+ [Como o Amazon Bedrock AgentCore oferece suporte ao MCP](#runtime-mcp-how-it-works)
+ [Pré-requisitos](#runtime-mcp-prerequisites)
+ [Etapa 1: Crie seu servidor MCP](#runtime-mcp-create-server)
+ [Etapa 2: Teste seu servidor MCP localmente](#runtime-mcp-test-locally)
+ [Etapa 3: implantar seu servidor MCP em AWS](#runtime-mcp-deploy-aws)
+ [Etapa 4: Invocar seu servidor MCP implantado](#runtime-mcp-invoke-server)
+ [Respostas de erro de autenticação para OAuth-Configured agentes](#runtime-mcp-auth-error-responses)
+ [Fluxo de ponta a ponta com Auth0](#runtime-mcp-auth0-flow)
+ [Apêndice](#runtime-mcp-appendix)

## Como o Amazon Bedrock AgentCore oferece suporte ao MCP
<a name="runtime-mcp-how-it-works"></a>

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 caminho`0.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](mcp-stateful-features.md).

A carga útil da [InvokeAgentRuntime](https://docs.aws.amazon.com/bedrock-agentcore/latest/APIReference/API_InvokeAgentRuntime.html)API é transmitida diretamente, permitindo que mensagens RPC de protocolos como MCP sejam facilmente enviadas por proxy.

## Pré-requisitos
<a name="runtime-mcp-prerequisites"></a>
+ 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
<a name="runtime-mcp-create-server"></a>

### Instalar os pacotes obrigatórios
<a name="runtime-mcp-install-packages"></a>

Primeiro, instale o pacote MCP:

```
pip install mcp
```

### Crie seu primeiro servidor MCP
<a name="runtime-mcp-create-first-server"></a>

Crie um novo arquivo chamado`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")
```

### Entendendo o código
<a name="runtime-mcp-code-explanation"></a>
+  **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](mcp-stateful-features.md).

## Etapa 2: Teste seu servidor MCP localmente
<a name="runtime-mcp-test-locally"></a>

### Inicie seu servidor MCP
<a name="runtime-mcp-start-server"></a>

Execute seu servidor MCP localmente:

```
python my_mcp_server.py
```

Você deve ver a saída indicando que o servidor está sendo executado na porta`8000`.

### Teste com o cliente MCP
<a name="runtime-mcp-test-client"></a>

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](#runtime-mcp-appendix-b) o inspetor MCP.

## Etapa 3: implantar seu servidor MCP em AWS
<a name="runtime-mcp-deploy-aws"></a>

### Instalar ferramentas de implantação
<a name="runtime-mcp-install-deployment-tools"></a>

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 chamado`requirements.txt`, adicione o seguinte a ele:

```
mcp
```

 `requirements.txt`especifica os requisitos que o agente precisa para implantação no AgentCore Runtime.

### Crie seu projeto para implantação
<a name="runtime-mcp-configure-deployment"></a>

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](#runtime-mcp-appendix-a) 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](service-linked-roles.md#identity-service-linked-role).

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
<a name="runtime-mcp-deploy"></a>

Implante seu agente:

```
agentcore deploy
```

Esse comando irá:

1. Package o código e as dependências do seu agente

1. Faça o upload do artefato de implantação para o Amazon S3

1. Crie um tempo de execução do Amazon Bedrock AgentCore 

1. 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
<a name="runtime-mcp-invoke-server"></a>

### Teste com o cliente MCP (remoto)
<a name="runtime-mcp-test-remote"></a>

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.](https://modelcontextprotocol.io/specification/2025-06-18/basic/transports#sending-messages-to-the-server) Os tipos de mídia aceitáveis sã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](#runtime-mcp-appendix-c) com o inspetor MCP.

## Respostas de erro de autenticação para OAuth-Configured agentes
<a name="runtime-mcp-auth-error-responses"></a>

OAuth-configured os agentes seguem os padrões de autenticação [RFC 6749 (OAuth 2.0](https://datatracker.ietf.org/doc/html/rfc6749)). Quando a autenticação está ausente, o serviço retorna uma resposta 401 não autorizada com um WWW-Authenticate cabeçalho (de acordo com a [RFC 7235](https://datatracker.ietf.org/doc/html/rfc7235)), permitindo que os clientes descubram os endpoints do servidor de autorização por meio da API. GetRuntimeProtectedResourceMetadata 

### 401 Não autorizado - Autenticação ausente
<a name="runtime-mcp-auth-401-unauthorized"></a>

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
<a name="runtime-mcp-auth0-flow"></a>

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
<a name="runtime-mcp-auth0-steps-1-3"></a>

Siga as etapas 1 a 3 da [Etapa 1: Criar seu servidor MCP](#runtime-mcp-create-server) até a [Etapa 3: Implantar seu servidor MCP AWS para](#runtime-mcp-deploy-aws) criar e testar seu servidor MCP.

### Etapa 4: criar o aplicativo Auth0
<a name="runtime-mcp-auth0-step-4"></a>

Siga as instruções de configuração do Auth0 em [Auth0](identity-idp-auth0.md) da Okta.

 **Ativar o registro dinâmico de clientes:** 

1. Painel → Configurações → Avançado

1. Alterne “Registro dinâmico de aplicativos OIDC” → ATIVADO

1. Salve as alterações.

Para obter mais informações, consulte a [documentação do Auth0 Dynamic Client Registration](https://auth0.com/docs/get-started/applications/dynamic-client-registration).

### Etapa 5: criar seu projeto para implantação
<a name="runtime-mcp-auth0-step-5"></a>

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
<a name="runtime-mcp-auth0-step-6"></a>

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
<a name="runtime-mcp-auth0-step-7"></a>

Esse cliente é baseado no exemplo [oficial do MCP SDK simple-auth-client](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/clients/simple-auth-client/mcp_simple_auth_client/main.py) com modificações. Auth0-specific 

**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](https://community.auth0.com/t/jwt-tokens-with-dynamic-application-registration/189741).

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](https://github.com/modelcontextprotocol/python-sdk/issues/1664) [e grupos de regras gerenciados pelo WAF](https://docs.aws.amazon.com/waf/latest/developerguide/aws-managed-rule-groups-baseline.html).

#### Código do cliente MCP Python Auth0
<a name="runtime-mcp-auth0-client-code"></a>

```
#!/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:

1. Defina as variáveis de ambiente necessárias:

   ```
   export AGENT_ARN="arn:aws:bedrock:us-west-2:123456789012:agent/ABCD1234"
   ```

1. Defina a variável de Auth0-specific ambiente (necessária somente para Auth0):

   ```
   export AUTH0_API_IDENTIFIER="your-api-identifier"
   ```

1. 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 `audience` parâ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
<a name="runtime-mcp-appendix"></a>

### Configurar o grupo de usuários do Cognito para autenticação
<a name="runtime-mcp-appendix-a"></a>

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 comando`source 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](runtime-oauth.md).

Depois de executar esse script, observe os seguintes valores para uso na configuração de implantação:
+ URL de descoberta: usado durante a `agentcore create` etapa
+ ID do cliente: usado durante a `agentcore create` etapa
+ Token do portador: usado ao invocar seu servidor implantado

### Teste local com o inspetor MCP
<a name="runtime-mcp-appendix-b"></a>

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:

1. Navegue até `http://localhost:6274` no seu navegador

1. Cole o URL do servidor MCP (`http://localhost:8000/mcp`) no campo de conexão do Inspector MCP

1. Você verá suas ferramentas listadas na barra lateral

1. Clique em qualquer ferramenta para testá-la

1. Preencha os parâmetros (por exemplo, para`add_numbers`, insira valores para `a` e`b`)

1. Clique em “Ferramenta de chamada” para ver o resultado

### Teste remoto com o inspetor MCP
<a name="runtime-mcp-appendix-c"></a>

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:

1. Inicie o Inspector MCP:

   ```
   npx @modelcontextprotocol/inspector
   ```

1. 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=DEFAULT
     ```

     Exemplo 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 `Authorization` e valor do cabeçalho `Bearer YOUR_TOKEN` 
   + Clique em “Connect”

1. Teste suas ferramentas da mesma forma que você fez localmente