Implementazione dei server MCP in Runtime AgentCore

Amazon Bedrock AgentCore Runtime consente di distribuire ed eseguire server Model Context Protocol (MCP) nel Runtime. AgentCore Questa guida ti guida attraverso la creazione, il test e la distribuzione del tuo primo server MCP.

Per vedere un esempio, consulta https://github.com/awslabs/amazon-bedrock-agentcore-samples/tree/main/01-tutorials/01-AgentCore-runtime/02-hosting-MCP-server.

In questa sezione, imparerai:

• Come creare un server MCP con strumenti

• Come testare il server localmente

• Come installare il server su AWS

• Come richiamare il server distribuito

Per ulteriori informazioni su MCP, vedere Contratto di protocollo MCP.

In che modo Amazon Bedrock AgentCore supporta MCP

Quando configuri un Amazon Bedrock AgentCore Runtime con il protocollo MCP, il servizio prevede che i contenitori del server MCP siano disponibili sul percorso0.0.0.0:8000/mcp, che è il percorso predefinito supportato dalla maggior parte degli SDK ufficiali per server MCP.

Amazon Bedrock AgentCore supporta server MCP HTTP Streamable sia stateless che stateful. Per impostazione predefinita, la modalità stateless () è consigliata per i server MCP di base. stateless_http=True La piattaforma aggiunge automaticamente un'Mcp-Session-Idintestazione per qualsiasi richiesta senza una, in modo che i client MCP possano mantenere la continuità della connessione alla stessa sessione di Amazon Bedrock Runtime AgentCore .

Per i server MCP che richiedono interazioni a più turni (elicitazione), LLM-generated contenuti (campionamento) o notifiche di avanzamento, la modalità stateful () abilita queste funzionalità. stateless_http=False In modalità stateful, il runtime conserva lo stato della sessione MCP tra le richieste all'interno della stessa chiamata. Per ulteriori informazioni, vedete Funzionalità del server Stateful MCP.

Il payload dell'InvokeAgentRuntimeAPI viene trasmesso direttamente, il che consente di inoltrare facilmente tramite proxy i messaggi RPC di protocolli come MCP.

Prerequisiti

• Python 3.10 o superiore installato e conoscenza di base di Python

• Un AWS account con le autorizzazioni appropriate e le credenziali locali configurate

Fase 1: Crea il tuo server MCP

Installazione dei pacchetti obbligatori

Innanzitutto, installa il pacchetto MCP:

pip install mcp

Crea il tuo primo server MCP

Crea un nuovo file chiamatomy_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")

Comprendere il codice

• FastMCP: crea un server MCP in grado di ospitare i tuoi strumenti

• @mcp .tool (): Decoratore che trasforma le funzioni Python in strumenti MCP

• Strumenti: tre semplici strumenti che illustrano diversi tipi di operazioni

• stateless_http=True: configura il server in modalità stateless, che è l'impostazione predefinita per i server MCP di base

Suggerimento

Per i server MCP che richiedono interazioni a più turni (elicitazione) o contenuti (campionamento), utilizzare per abilitare la modalità stateful. LLM-generated stateless_http=False I server MCP Stateful mantengono il contesto della sessione tra più richieste all'interno della stessa invocazione dello strumento. Per ulteriori informazioni, vedete Funzionalità del server Stateful MCP.

Passaggio 2: testate il server MCP localmente

Avvia il tuo server MCP

Esegui il tuo server MCP localmente:

python my_mcp_server.py

Dovresti vedere un output che indica che il server è in esecuzione sulla porta8000.

Esegui il test con il client MCP

Da un nuovo terminale, crea un nuovo file my_mcp_client.py ed eseguilo utilizzando 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())

È inoltre possibile testare il server utilizzando MCP Inspector come descritto in Test locali con MCP Inspector.

Fase 3: Implementate il server MCP su AWS

Installa gli strumenti di distribuzione

Installa la AgentCore CLI:

npm install -g @aws/agentcore

Utilizzi la AgentCore CLI per distribuire il tuo agente in Runtime. AgentCore

Crea una cartella di progetto con la seguente struttura:

## 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

Crea un nuovo file chiamatorequirements.txt, aggiungi quanto segue:

mcp

requirements.txtspecifica i requisiti necessari all'agente per la distribuzione in AgentCore Runtime.

Crea il tuo progetto per la distribuzione

Prima di creare il progetto, devi configurare un pool di utenti Cognito per l'autenticazione, come descritto in Configurare il pool di utenti Cognito per l'autenticazione. Ciò fornisce i token OAuth necessari per un accesso sicuro al server distribuito.

Nota

A partire dal 7 ottobre 2025, Amazon Bedrock AgentCore utilizza un Service-Linked ruolo per le autorizzazioni relative all'identità del carico di lavoro quando si utilizza l'autenticazione OAuth. Per informazioni dettagliate su questa modifica, consulta Identity service-linked role.

Dopo aver impostato l'autenticazione, crea un nuovo progetto con il protocollo MCP:

agentcore create --protocol MCP

Segui le istruzioni interattive per fornire un nome al progetto. La CLI supporta la struttura del progetto, incluso un file di configurazione. agentcore/agentcore.json Copia il my_mcp_server.py file nella directory del codice agente del progetto generato e assicurati che il punto di ingresso agentcore/agentcore.json punti al file del server.

Distribuisci su AWS

Implementa il tuo agente:

agentcore deploy

Questo comando consentirà di:

1. Package del codice dell'agente e delle dipendenze

2. Carica l'elemento di distribuzione su Amazon S3

3. Crea un runtime Amazon Bedrock AgentCore

4. Implementa il tuo agente su AWS

Dopo la distribuzione, riceverai un ARN di runtime dell'agente simile a:

arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_mcp_server-xyz123

Passaggio 4: richiama il server MCP distribuito

Esegui il test con il client MCP (remoto)

Prima del test, impostate le seguenti variabili di ambiente:

• Arn dell'agente di esportazione come variabile di ambiente: export AGENT_ARN="agent_arn"

• Esporta il token bearer come variabile di ambiente: export BEARER_TOKEN="bearer_token"

se inserite un'Acceptintestazione, questa deve seguire lo standard MCP. I tipi di file multimediali accettabili sono application/json e. text/event-stream

Crea un nuovo file my_mcp_client_remote.py ed eseguilo utilizzando 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())

È inoltre possibile testare il server distribuito utilizzando MCP Inspector, come descritto in Test remoto con MCP Inspector.

Risposte agli errori di autenticazione per gli agenti OAuth-Configured

OAuth-configured gli agenti seguono gli standard di autenticazione RFC 6749 (OAuth 2.0). Quando manca l'autenticazione, il servizio restituisce una risposta 401 Unauthorized con un' WWW-Authenticate intestazione (secondo RFC 7235), che consente ai client di scoprire gli endpoint del server di autorizzazione tramite l'API. GetRuntimeProtectedResourceMetadata

401 Non autorizzato: autenticazione mancante

Quando non viene fornito alcun token Bearer nell'intestazione di autorizzazione, la risposta è:

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}"

Flusso end-to-end con Auth0

Questa sezione dimostra l'autenticazione OAuth utilizzando Auth0 come provider di identità. Utilizziamo Auth0 per questo esempio perché supporta Dynamic Client Registration (DCR), che semplifica il processo di configurazione del client consentendo ai client di registrarsi a livello di codice in fase di esecuzione.

Fase 1 - Fase 3: Creare e testare il server MCP

Segui i passaggi 1-3 dal Passaggio 1: Creazione del server MCP al Passaggio 3: Implementa il server MCP per creare e AWS testare il server MCP.

Fase 4: Creazione dell'applicazione Auth0

Segui le istruzioni di configurazione Auth0 su Auth0 di Okta.

Abilita la registrazione dinamica del client:

1. Dashboard → Impostazioni → Avanzate

2. Attiva «Registrazione dinamica dell'applicazione OIDC» →

3. Salva le modifiche.

Per ulteriori informazioni, consulta la documentazione Auth0 Dynamic Client Registration.

Fase 5: Crea il tuo progetto per la distribuzione

Dopo aver impostato l'autenticazione, crea un nuovo progetto con il protocollo MCP:

agentcore create --protocol MCP

Segui le istruzioni interattive per fornire un nome al progetto. La CLI supporta la struttura del progetto, incluso un file di configurazione. agentcore/agentcore.json Copia il my_mcp_server.py file nella directory del codice agente del progetto generato e assicurati che il punto di ingresso agentcore/agentcore.json punti al file del server.

Fase 6: Esegui la distribuzione su AWS

Implementa il tuo agente:

agentcore deploy

Questo comando consentirà di:

• Package del codice dell'agente e delle dipendenze

• Carica l'elemento di distribuzione su Amazon S3

• Crea un runtime Amazon Bedrock AgentCore

• Implementa il tuo agente su AWS

Dopo la distribuzione, riceverai un ARN di runtime dell'agente simile a:

arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_mcp_server-xyz123

Passaggio 7: richiama l'agente distribuito

Questo client si basa sull'esempio ufficiale di MCP SDK simple-auth-client con modifiche. Auth0-specific

Nota

Quando si utilizza Auth0 con Dynamic Client Registration, è necessario includere il audience parametro nelle richieste di autorizzazione per ricevere i token JWT. Senza questo parametro, Auth0 restituisce token opachi o token JWE (crittografati) anziché token JWT standard. L'SDK MCP invia il parametro OAuth 2.0 (RFC 8707), ma Auth0 richiede il resource parametro OIDC per i token JWT. audience Entrambi i parametri hanno scopi audience simili, ma Auth0 dà la priorità. Per ulteriori informazioni, consulta Auth0 Community - JWT tokens with Dynamic Application Registration.

Crea un file denominato mcp_auth0_client.py con il codice seguente. Questo client gestisce Auth0-specific i requisiti, incluso il parametro audience:

Nota

Il codice include l'applicazione di patch httpx per inserire User-Agent intestazioni in tutte le richieste HTTP. Ciò è necessario perché l'SDK MCP Python attualmente non User-Agent include le intestazioni nelle sue richieste HTTP, il che può causare problemi AWS con le regole WAF che richiedono intestazioni. User-Agent Per ulteriori informazioni, consulta MCP Python SDK Issue #1664 AWS e WAF managed rule groups.

Codice client MCP Python Auth0

#!/usr/bin/env python3
"""
MCP client with OAuth authentication support for Auth0.

Based on the official MCP SDK simple-auth-client example with Auth0 compatibility.
Adds support for Auth0's 'audience' parameter requirement.

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

    # Required for Auth0
    export AUTH0_API_IDENTIFIER="your-api-identifier"

    # Optional - custom endpoint for beta/dev environments
    export CUSTOM_ENDPOINT="https://beta.example.com"

    python mcp_auth0_client.py

The client will automatically:
- Encode the Agent ARN for use in the URL
- Construct the MCP invocation endpoint URL
- Add Auth0 'audience' parameter to authorization requests (when using Auth0)
- Work with any OAuth 2.0 compliant identity provider
"""

import asyncio
import httpx
import os
import threading
import time
import webbrowser
from datetime import timedelta
from http.server import BaseHTTPRequestHandler, HTTPServer
from typing import Any
from urllib.parse import parse_qs, urlencode, urlparse, urlunparse

# Patch httpx at the request level to inject User-Agent header
# This ensures ALL HTTP requests have the User-Agent header, including OAuth discovery calls
_original_httpx_request = httpx.Request.__init__

def _patched_httpx_request_init(self, method, url, *args, **kwargs):
    """Patched Request.__init__ that injects User-Agent header into all HTTP requests."""
    # Get or create headers
    headers = kwargs.get('headers')
    if headers is None:
        headers = {}
        kwargs['headers'] = headers

    # Convert to mutable dict if needed
    if not isinstance(headers, dict):
        headers = dict(headers)
        kwargs['headers'] = headers

    # Inject User-Agent if not present (case-insensitive check)
    if 'User-Agent' not in headers and 'user-agent' not in headers:
        headers['User-Agent'] = 'python-mcp-sdk/1.0 (BedrockAgentCore-Runtime)'

    # Call original __init__
    _original_httpx_request(self, method, url, *args, **kwargs)

# Apply the patch globally before importing MCP modules
httpx.Request.__init__ = _patched_httpx_request_init

# Now import MCP modules - they will use patched httpx
from mcp.client.auth import OAuthClientProvider, TokenStorage
from mcp.client.session import ClientSession
from mcp.client.sse import sse_client
from mcp.client.streamable_http import streamablehttp_client
from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken

class InMemoryTokenStorage(TokenStorage):
    """Simple in-memory token storage implementation."""

    def __init__(self):
        self._tokens: OAuthToken | None = None
        self._client_info: OAuthClientInformationFull | None = None

    async def get_tokens(self) -> OAuthToken | None:
        return self._tokens

    async def set_tokens(self, tokens: OAuthToken) -> None:
        self._tokens = tokens

    async def get_client_info(self) -> OAuthClientInformationFull | None:
        return self._client_info

    async def set_client_info(self, client_info: OAuthClientInformationFull) -> None:
        self._client_info = client_info

class CallbackHandler(BaseHTTPRequestHandler):
    """Simple HTTP handler to capture OAuth callback."""

    def __init__(self, request, client_address, server, callback_data):
        """Initialize with callback data storage."""
        self.callback_data = callback_data
        super().__init__(request, client_address, server)

    def do_GET(self):
        """Handle GET request from OAuth redirect."""
        parsed = urlparse(self.path)
        query_params = parse_qs(parsed.query)

        if "code" in query_params:
            self.callback_data["authorization_code"] = query_params["code"][0]
            self.callback_data["state"] = query_params.get("state", [None])[0]
            self.send_response(200)
            self.send_header("Content-type", "text/html")
            self.end_headers()
            self.wfile.write(b"""
            <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()

Per utilizzare il client:

1. Imposta le variabili di ambiente richieste:

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

2. Imposta la variabile di Auth0-specific ambiente (richiesta solo per Auth0):

   export AUTH0_API_IDENTIFIER="your-api-identifier"

3. Esegui il client:

   python mcp_auth0_client.py

Il client eseguirà automaticamente:

• Codifica l'ARN dell'agente per utilizzarlo nell'URL

• Costruisci l'URL dell'endpoint di invocazione MCP

• Aggiungi il audience parametro Auth0 alle richieste di autorizzazione (quando usi Auth0)

• Funziona con qualsiasi provider di identità conforme a OAuth 2.0

Appendice

Configura il pool di utenti di Cognito per l'autenticazione

Crea un nuovo file setup_cognito.sh e aggiungi il seguente contenuto.

#!/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"

Apri una finestra di terminale e imposta le seguenti variabili di ambiente:

• REGION— la AWS regione che vuoi usare

• USERNAME— il nome utente per il nuovo utente

• PASSWORD— la password per il nuovo utente

export REGION=us-east-1 // set your desired Region
export USERNAME=USER NAME
export PASSWORD=PASSWORD

Esegui lo script utilizzando il comandosource setup_cognito.sh.

Nota

Per informazioni dettagliate sulla configurazione dell'autenticazione OAuth e sul Service-Linked ruolo, consulta Autenticazione e autorizzazione con autenticazione in entrata e autenticazione in uscita.

Dopo aver eseguito questo script, prendi nota dei seguenti valori da utilizzare nella configurazione di distribuzione:

• Discovery URL: utilizzato durante la agentcore create fase

• ID client: utilizzato durante la agentcore create fase

• Token Bearer: utilizzato per richiamare il server distribuito

Test locali con MCP Inspector

MCP Inspector è uno strumento visivo per testare i server MCP. Per utilizzarlo, è necessario:

• Node.js e npm installato

Installa ed esegui MCP Inspector:

npx @modelcontextprotocol/inspector

Ciò consentirà di:

• Avviare il server MCP Inspector

• Visualizzate un URL nel terminale (in genere) http://localhost:6274

Per utilizzare l'Inspector:

1. Accedi a http://localhost:6274 nel tuo browser

2. Incolla l'URL del server MCP (http://localhost:8000/mcp) nel campo di connessione MCP Inspector

3. Vedrai i tuoi strumenti elencati nella barra laterale

4. Fai clic su uno strumento per testarlo

5. Inserisci i parametri (ad esempio, peradd_numbers, inserisci i valori per a eb)

6. Fai clic su «Call Tool» per vedere il risultato

Test a distanza con MCP Inspector

È inoltre possibile testare il server distribuito utilizzando MCP Inspector. Innanzitutto, l'ARN del URL-encode tuo agente:

export AGENT_ARN="arn:aws:bedrock-agentcore:us-west-2:123456789012:runtime/my_mcp_server-xyz123"
echo -n $AGENT_ARN | jq -sRr '@uri'

Questo restituisce l' URL-encoded ARN:

arn%3Aaws%3Abedrock-agentcore%3Aus-west-2%3A123456789012%3Aruntime%2Fmy_mcp_server-xyz123

Quindi connettiti con MCP Inspector:

1. Avvia MCP Inspector:

   npx @modelcontextprotocol/inspector

2. Nell'interfaccia web:

   • Seleziona «Streamable HTTP» come mezzo di trasporto

   • Inserisci l'URL dell'endpoint del tuo agente utilizzando l'ARN codificato. Assicurati di utilizzare la stessa regione dell'ARN del tuo agente:

     https://bedrock-agentcore.REGION.amazonaws.com/runtimes/ENCODED_ARN/invocations?qualifier=DEFAULT

     Esempio per 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

   • Aggiungi il tuo token Bearer nella sezione Autenticazione con il nome e il valore dell'intestazione Authorization Bearer YOUR_TOKEN

   • Fai clic su «Connect»

3. Testa i tuoi strumenti proprio come hai fatto localmente