Menyebarkan server MCP di Runtime AgentCore
Amazon Bedrock AgentCore Runtime memungkinkan Anda menerapkan dan menjalankan server Model Context Protocol (MCP) di Runtime. AgentCore Panduan ini memandu Anda melalui pembuatan, pengujian, dan penerapan server MCP pertama Anda.
Sebagai contoh, lihat https://github.com/awslabs/amazon-bedrock-agentcore-samples/tree/main/01-tutorials/01-AgentCore-runtime/02-hosting-MCP-server
Di bagian ini, Anda belajar:
-
Cara membuat server MCP dengan alat
-
Cara menguji server Anda secara lokal
-
Cara menyebarkan server Anda ke AWS
-
Cara memanggil server yang Anda gunakan
Untuk informasi selengkapnya tentang MCP, lihat kontrak protokol MCP.
Topik
Bagaimana Amazon Bedrock AgentCore mendukung MCP
Saat Anda mengonfigurasi AgentCore Runtime Amazon Bedrock dengan protokol MCP, layanan mengharapkan kontainer server MCP tersedia di jalur0.0.0.0:8000/mcp, yang merupakan jalur default yang didukung oleh sebagian besar SDK server MCP resmi.
Amazon Bedrock AgentCore mendukung server MCP Streamable-HTTP stateless dan stateful. Secara default, stateless mode (stateless_http=True) direkomendasikan untuk server MCP dasar. Platform secara otomatis menambahkan Mcp-Session-Id header untuk permintaan apa pun tanpa satu, sehingga klien MCP dapat mempertahankan kontinuitas koneksi ke sesi Runtime Amazon Bedrock AgentCore yang sama.
Untuk server MCP yang memerlukan interaksi multi-putaran (elicitation), LLM-generated konten (sampling), atau pemberitahuan kemajuan, stateful mode () memungkinkan kemampuan ini. stateless_http=False Dalam mode stateful, runtime mempertahankan status sesi MCP di seluruh permintaan dalam pemanggilan yang sama. Untuk informasi selengkapnya, lihat Fitur server MCP stateful.
Muatan InvokeAgentRuntimeAPI dilewatkan secara langsung, memungkinkan pesan protokol RPC seperti MCP mudah diproksi.
Prasyarat
-
Python 3.10 atau lebih tinggi diinstal dan pemahaman dasar Python
-
AWS Akun dengan izin yang sesuai dan kredensi lokal yang dikonfigurasi
Langkah 1: Buat server MCP Anda
Menginstal paket yang diperlukan
Pertama, instal paket MCP:
pip install mcp
Buat server MCP pertama Anda
Buat file baru bernamamy_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")
Memahami kode
-
FastMCP: Membuat server MCP yang dapat meng-host alat Anda
-
@mcp .tool (): Dekorator yang mengubah fungsi Python Anda menjadi alat MCP
-
Alat: Tiga alat sederhana yang menunjukkan berbagai jenis operasi
-
stateless_http=True: Mengkonfigurasi server dalam mode stateless, yang merupakan default untuk server MCP dasar
Tip
Untuk server MCP yang memerlukan interaksi multi-putaran (elicitation) atau LLM-generated content (sampling), gunakan stateless_http=False untuk mengaktifkan mode stateful. Server MCP stateful mempertahankan konteks sesi di beberapa permintaan dalam pemanggilan alat yang sama. Untuk informasi selengkapnya, lihat Fitur server MCP stateful.
Langkah 2: Uji server MCP Anda secara lokal
Mulai server MCP Anda
Jalankan server MCP Anda secara lokal:
python my_mcp_server.py
Anda akan melihat output yang menunjukkan server berjalan pada port8000.
Uji dengan klien MCP
Dari terminal baru, buat file baru my_mcp_client.py dan jalankan menggunakan 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())
Anda juga dapat menguji server Anda menggunakan MCP Inspector seperti yang dijelaskan dalam pengujian Lokal dengan inspektur MCP.
Langkah 3: Menyebarkan server MCP Anda ke AWS
Instal alat penyebaran
Instal AgentCore CLI:
npm install -g @aws/agentcore
Anda menggunakan AgentCore CLI untuk menyebarkan agen Anda ke Runtime. AgentCore
Buat folder proyek dengan struktur berikut:
## 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
Buat file baru bernamarequirements.txt, tambahkan yang berikut ini:
mcp
requirements.txtmenentukan persyaratan yang dibutuhkan agen untuk penyebaran ke AgentCore Runtime.
Buat proyek Anda untuk penerapan
Sebelum membuat project, Anda perlu menyiapkan kumpulan pengguna Cognito untuk autentikasi seperti yang dijelaskan dalam Mengatur kumpulan pengguna Cognito untuk autentikasi. Ini menyediakan token OAuth yang diperlukan untuk akses aman ke server yang Anda gunakan.
catatan
Mulai 7 Oktober 2025, Amazon Bedrock AgentCore menggunakan Service-Linked Peran untuk izin identitas beban kerja saat menggunakan otentikasi OAuth. Untuk informasi mendetail tentang perubahan ini, lihat Peran terkait layanan identitas.
Setelah menyiapkan otentikasi, perancah proyek baru dengan protokol MCP:
agentcore create --protocol MCP
Ikuti petunjuk interaktif untuk memberikan nama proyek. CLI merancah struktur proyek termasuk file konfigurasi. agentcore/agentcore.json Salin my_mcp_server.py file Anda ke direktori kode agen proyek yang dihasilkan, dan pastikan titik masuk agentcore/agentcore.json menunjuk ke file server Anda.
Menyebarkan ke AWS
Menyebarkan agen Anda:
agentcore deploy
Perintah ini akan:
-
Package kode agen dan dependensi Anda
-
Unggah artefak penerapan ke Amazon S3
-
Buat runtime Amazon Bedrock AgentCore
-
Menyebarkan agen Anda ke AWS
Setelah penerapan, Anda akan menerima ARN runtime agen yang terlihat seperti:
arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_mcp_server-xyz123
Langkah 4: Panggil server MCP yang Anda gunakan
Uji dengan klien MCP (jarak jauh)
Sebelum pengujian, atur variabel lingkungan berikut:
-
Agen ekspor ARN sebagai variabel lingkungan:
export AGENT_ARN="agent_arn" -
Ekspor token pembawa sebagai variabel lingkungan:
export BEARER_TOKEN="bearer_token"
jika Anda memasukkan Accept header, itu harus mengikuti standar MCPapplication/json dantext/event-stream.
Buat file baru my_mcp_client_remote.py dan jalankan menggunakan 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())
Anda juga dapat menguji server yang digunakan menggunakan MCP Inspector seperti yang dijelaskan dalam Pengujian jarak jauh dengan inspektur MCP.
Tanggapan Kesalahan Otentikasi untuk Agen OAuth-Configured
OAuth-configured agen mengikuti standar otentikasi RFC 6749 (OAuth 2.0
401 Tidak Sah - Otentikasi Hilang
Ketika tidak ada token Bearer yang disediakan di header Otorisasi, responsnya adalah:
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}"
End to End Flow dengan Auth0
Bagian ini menunjukkan otentikasi OAuth menggunakan Auth0 sebagai penyedia identitas. Kami menggunakan Auth0 untuk contoh ini karena mendukung Dynamic Client Registration (DCR), yang menyederhanakan proses penyiapan klien dengan memungkinkan klien mendaftarkan diri secara terprogram saat runtime.
Langkah 1 - Langkah 3: Buat dan uji server MCP Anda
Ikuti Langkah 1-3 dari Langkah 1: Buat server MCP Anda melalui Langkah 3: Menyebarkan server MCP Anda AWS untuk membuat dan menguji server MCP Anda.
Langkah 4: Buat aplikasi Auth0
Ikuti instruksi pengaturan Auth0 di Auth0 oleh Okta.
Aktifkan Pendaftaran Klien Dinamis:
-
Dasbor → Pengaturan → Lanjutan
-
Alihkan “Pendaftaran Aplikasi Dinamis OIDC” → AKTIF
-
Simpan perubahan
Untuk informasi selengkapnya, lihat dokumentasi Pendaftaran Klien Dinamis Auth0
Langkah 5: Buat proyek Anda untuk penyebaran
Setelah menyiapkan otentikasi, perancah proyek baru dengan protokol MCP:
agentcore create --protocol MCP
Ikuti petunjuk interaktif untuk memberikan nama proyek. CLI merancah struktur proyek termasuk file konfigurasi. agentcore/agentcore.json Salin my_mcp_server.py file Anda ke direktori kode agen proyek yang dihasilkan, dan pastikan titik masuk agentcore/agentcore.json menunjuk ke file server Anda.
Langkah 6: Menyebarkan ke AWS
Menyebarkan agen Anda:
agentcore deploy
Perintah ini akan:
-
Package kode agen dan dependensi Anda
-
Unggah artefak penerapan ke Amazon S3
-
Buat runtime Amazon Bedrock AgentCore
-
Menyebarkan agen Anda ke AWS
Setelah penerapan, Anda akan menerima ARN runtime agen yang terlihat seperti:
arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_mcp_server-xyz123
Langkah 7: Panggil Agen yang Anda gunakan
Klien ini didasarkan pada contoh MCP SDK simple-auth-client resmi
catatan
Saat menggunakan Auth0 dengan Pendaftaran Klien Dinamis, Anda harus menyertakan audience parameter dalam permintaan otorisasi untuk menerima token JWT. Tanpa parameter ini, Auth0 mengembalikan token buram atau token JWE (terenkripsi) alih-alih token JWT standar. MCP SDK mengirimkan parameter OAuth 2.0 (RFC 8707), tetapi Auth0 memerlukan resource parameter OIDC untuk token JWT. audience Kedua parameter melayani tujuan yang sama tetapi Auth0 memprioritaskanaudience. Untuk informasi selengkapnya, lihat Komunitas Auth0 - Token JWT dengan Pendaftaran Aplikasi Dinamis
Buat file bernama mcp_auth0_client.py dengan kode berikut. Klien ini menangani Auth0-specific persyaratan termasuk parameter audiens:
catatan
Kode ini mencakup patch httpx untuk menyuntikkan User-Agent header ke semua permintaan HTTP. Ini diperlukan karena MCP Python SDK saat ini tidak User-Agent menyertakan header dalam permintaan HTTP-nya, yang dapat menyebabkan masalah AWS dengan aturan WAF yang memerlukan header. User-Agent Untuk informasi selengkapnya, lihat MCP Python SDK Issue
#!/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()
Untuk menggunakan klien:
-
Tetapkan variabel lingkungan yang diperlukan:
export AGENT_ARN="arn:aws:bedrock:us-west-2:123456789012:agent/ABCD1234" -
Setel variabel Auth0-specific lingkungan (hanya diperlukan untuk Auth0):
export AUTH0_API_IDENTIFIER="your-api-identifier" -
Jalankan klien:
python mcp_auth0_client.py
Klien akan secara otomatis:
-
Encode Agen ARN untuk digunakan di URL
-
Membangun URL endpoint pemanggilan MCP
-
Tambahkan
audienceparameter Auth0 ke permintaan otorisasi (saat menggunakan Auth0) -
Bekerja dengan penyedia identitas yang sesuai dengan OAuth 2.0
Lampiran
Siapkan kumpulan pengguna Cognito untuk otentikasi
Buat file baru setup_cognito.sh dan tambahkan konten berikut.
#!/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"
Buka jendela terminal dan atur variabel lingkungan berikut:
-
REGION— AWS Wilayah yang ingin Anda gunakan -
USERNAME— nama pengguna untuk pengguna baru -
PASSWORD— kata sandi untuk pengguna baru
export REGION=us-east-1 // set your desired Region export USERNAME=USER NAME export PASSWORD=PASSWORD
Jalankan skrip menggunakan perintahsource setup_cognito.sh.
catatan
Untuk mengetahui detail pengaturan otentikasi OAuth dan informasi Service-Linked Peran, lihat Mengautentikasi dan mengotorisasi dengan Auth Masuk dan Auth Keluar.
Setelah menjalankan skrip ini, perhatikan nilai berikut untuk digunakan dalam konfigurasi penerapan:
-
URL Penemuan: Digunakan selama
agentcore createlangkah -
ID Klien: Digunakan selama
agentcore createlangkah -
Token Pembawa: Digunakan saat menjalankan server yang Anda gunakan
Pengujian lokal dengan inspektur MCP
MCP Inspector adalah alat visual untuk menguji server MCP. Untuk menggunakannya, Anda perlu:
-
Node.js dan npm diinstal
Instal dan jalankan MCP Inspector:
npx @modelcontextprotocol/inspector
Ini akan:
-
Mulai server MCP Inspector
-
Menampilkan URL di terminal Anda (biasanya
http://localhost:6274)
Untuk menggunakan Inspector:
-
Arahkan ke
http://localhost:6274di browser Anda -
Tempelkan URL server MCP (
http://localhost:8000/mcp) ke bidang koneksi MCP Inspector -
Anda akan melihat alat Anda tercantum di sidebar
-
Klik pada alat apa saja untuk mengujinya
-
Isi parameter (misalnya, untuk
add_numbers, masukkan nilai untukadanb) -
Klik “Alat Panggilan” untuk melihat hasilnya
Pengujian jarak jauh dengan inspektur MCP
Anda juga dapat menguji server yang digunakan menggunakan MCP Inspector. Pertama, agen ARN URL-encode Anda:
export AGENT_ARN="arn:aws:bedrock-agentcore:us-west-2:123456789012:runtime/my_mcp_server-xyz123" echo -n $AGENT_ARN | jq -sRr '@uri'
Ini menghasilkan URL-encoded ARN:
arn%3Aaws%3Abedrock-agentcore%3Aus-west-2%3A123456789012%3Aruntime%2Fmy_mcp_server-xyz123
Kemudian terhubung dengan MCP Inspector:
-
Mulai MCP Inspector:
npx @modelcontextprotocol/inspector -
Di antarmuka web:
-
Pilih “Streamable HTTP” sebagai transportasi
-
Masukkan URL titik akhir agen Anda menggunakan ARN yang dikodekan. Pastikan untuk menggunakan wilayah yang sama dengan ARN agen Anda:
https://bedrock-agentcore.REGION.amazonaws.com/runtimes/ENCODED_ARN/invocations?qualifier=DEFAULTContoh untuk 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 -
Tambahkan token Bearer Anda di bagian Otentikasi dengan nama
Authorizationdan nilai headerBearer YOUR_TOKEN -
Klik “Connect”
-
-
Uji alat Anda seperti yang Anda lakukan secara lokal
