在 AgentCore 執行期中部署 MCP 伺服器
Amazon Bedrock AgentCore 執行期可讓您在 AgentCore 執行期中部署和執行模型內容通訊協定 (MCP) 伺服器。本指南會逐步引導您建立、測試和部署您的第一個 MCP 伺服器。
在本區段，您會學習：
-
如何使用 工具建立 MCP 伺服器
-
如何在本機測試您的伺服器
-
如何將伺服器部署至 AWS
-
如何叫用已部署的伺服器
如需 MCP 的詳細資訊，請參閱 MCP 通訊協定合約。
主題
Amazon Bedrock AgentCore 如何支援 MCP
當您使用 MCP 通訊協定設定 Amazon Bedrock AgentCore 執行期時，服務預期 MCP 伺服器容器可在路徑 0.0.0.0:8000/mcp 中使用，這是大多數官方 MCP 伺服器 SDKs 支援的預設路徑。
Amazon Bedrock AgentCore 支援無狀態和具狀態可串流 HTTP MCP 伺服器。根據預設，基本 MCP 伺服器建議使用無狀態模式 stateless_http=True ()。平台會自動為沒有 的任何請求新增 Mcp-Session-Id標頭，因此 MCP 用戶端可以維持與相同 Amazon Bedrock AgentCore 執行期工作階段的連線持續性。
對於需要多迴轉互動 （引出）、LLM 產生的內容 （取樣） 或進度通知的 MCP 伺服器，具狀態模式 stateless_http=False () 會啟用這些功能。在具狀態模式中，執行時間會保留相同調用內請求之間的 MCP 工作階段狀態。如需詳細資訊，請參閱具狀態 MCP 伺服器功能。
InvokeAgentRuntime API 的承載會直接傳遞，以便輕鬆代理 MCP 等通訊協定的 RPC 訊息。
先決條件
-
已安裝 Python 3.10 或更新版本並基本了解 Python
-
已設定適當許可和本機登入資料的 AWS 帳戶
步驟 1：建立 MCP 伺服器
安裝必要套件
首先，安裝 MCP 套件：
pip install mcp
建立您的第一個 MCP 伺服器
建立名為 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")
了解程式碼
-
FastMCP ：建立可託管您工具的 MCP 伺服器
-
@mcp.tool()：將您的 Python 函數轉換為 MCP 工具的裝飾器
-
工具 ：三種示範不同操作類型的簡單工具
-
stateless_http=True ：以無狀態模式設定伺服器，這是基本 MCP 伺服器的預設值
提示
對於需要多迴轉互動 （引出） 或 LLM 產生內容 （取樣） 的 MCP 伺服器，請使用 stateless_http=False 來啟用具狀態模式。具狀態 MCP 伺服器會維護相同工具調用中多個請求的工作階段內容。如需詳細資訊，請參閱具狀態 MCP 伺服器功能。
步驟 2：在本機測試 MCP 伺服器
啟動 MCP 伺服器
在本機執行 MCP 伺服器：
python my_mcp_server.py
您應該會看到輸出，指出伺服器正在連接埠 上執行8000。
使用 MCP 用戶端進行測試
從新的終端機，建立新的檔案my_mcp_client.py並使用 執行 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())
您也可以使用 MCP Inspector 來測試伺服器，如使用 MCP 檢查器的本機測試中所述。
步驟 3：將 MCP 伺服器部署至 AWS
安裝部署工具
安裝 AgentCore CLI：
npm install -g @aws/agentcore
您可以使用 AgentCore CLI 將代理程式部署到 AgentCore 執行期。
使用下列結構建立專案資料夾：
## 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
建立名為 requirements.txt 的新檔案，新增以下內容：
mcp
requirements.txt 指定代理程式部署到 AgentCore 執行期所需的需求。
建立您的專案以進行部署
在建立專案之前，您需要設定用於身分驗證的 Cognito 使用者集區，如設定用於身分驗證的 Cognito 使用者集區中所述。這可提供安全存取已部署伺服器所需的 OAuth 權杖。
注意
從 2025 年 10 月 7 日開始，Amazon Bedrock AgentCore 在使用 OAuth 身分驗證時，會針對工作負載身分許可使用服務連結角色。如需此變更的詳細資訊，請參閱身分服務連結角色。
設定身分驗證之後，請使用 MCP 通訊協定來堆疊新專案：
agentcore create --protocol MCP
依照互動式提示提供專案名稱。CLI 會堆疊包含agentcore/agentcore.json組態檔案的專案結構。將您的my_mcp_server.py檔案複製到產生的專案代理程式程式碼目錄，並確保進入點agentcore/agentcore.json指向您的伺服器檔案。
部署至 AWS
部署您的代理程式：
agentcore deploy
此命令將：
-
封裝您的代理程式程式碼和相依性
-
將部署成品上傳至 Amazon S3
-
建立 Amazon Bedrock AgentCore 執行期
-
將您的代理程式部署到 AWS
部署之後，您會收到客服人員執行期 ARN，如下所示：
arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_mcp_server-xyz123
步驟 4：叫用您部署的 MCP 伺服器
使用 MCP 用戶端進行測試 （遠端）
在測試之前，請設定下列環境變數：
-
匯出代理程式 ARN 做為環境變數：
export AGENT_ARN="agent_arn" -
匯出承載字符做為環境變數：
export BEARER_TOKEN="bearer_token"
如果您傳入 Accept標頭，則必須遵循 MCPapplication/json和 text/event-stream。
建立新的檔案my_mcp_client_remote.py並使用 執行 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())
您也可以使用 MCP Inspector 測試已部署的伺服器，如使用 MCP 檢查器的遠端測試中所述。
OAuth 設定代理程式的身分驗證錯誤回應
OAuth 設定的代理程式遵循 RFC 6749 (OAuth 2.0)
401 未授權 - 缺少身分驗證
當授權標頭中未提供承載字符時，回應為：
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}"
使用 Auth0 進行端對端流程
本節示範使用 Auth0 做為身分提供者的 OAuth 身分驗證。 Auth0 我們在此範例中使用 Auth0，因為它支援動態用戶端註冊 (DCR) ，它允許用戶端在執行時間以程式設計方式註冊自己，以簡化用戶端設定程序。
步驟 1 - 步驟 3：建立和測試 MCP 伺服器
請遵循步驟 1：透過步驟 3：將 MCP 伺服器部署至 來建立和測試 MCP 伺服器中的步驟 1-3 AWS。
步驟 4：建立 Auth0 應用程式
遵循 Okta 的 Auth0 中的 Auth0 設定指示。 Auth0
啟用動態用戶端註冊：
-
儀表板 → 設定 → 進階
-
切換「OIDC 動態應用程式註冊」→ ON
-
儲存變更
如需詳細資訊，請參閱 Auth0 動態用戶端註冊文件
步驟 5：建立您的專案以進行部署
設定身分驗證之後，請使用 MCP 通訊協定來堆疊新專案：
agentcore create --protocol MCP
依照互動式提示提供專案名稱。CLI 會堆疊包含agentcore/agentcore.json組態檔案的專案結構。將您的my_mcp_server.py檔案複製到產生的專案代理程式程式碼目錄，並確保進入點agentcore/agentcore.json指向您的伺服器檔案。
步驟 6：部署至 AWS
部署您的代理程式：
agentcore deploy
此命令將：
-
封裝您的代理程式程式碼和相依性
-
將部署成品上傳至 Amazon S3
-
建立 Amazon Bedrock AgentCore 執行期
-
將您的代理程式部署到 AWS
部署之後，您會收到客服人員執行期 ARN，如下所示：
arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_mcp_server-xyz123
步驟 7：叫用您部署的代理程式
此用戶端是以具有 Auth0-specific修改的官方 MCP SDK simple-auth-client範例
注意
搭配動態用戶端註冊使用 Auth0 時，您必須在授權請求中包含 audience 參數，才能接收 JWT 字符。如果沒有此參數，Auth0 會傳回不透明字符或 JWE （加密） 字符，而不是標準 JWT 字符。MCP SDK 會傳送 OAuth 2.0 的 resource 參數 (RFC 8707)，但 Auth0 需要 JWT 權杖的 OIDC audience 參數。這兩個參數都有類似的用途，但 Auth0 會優先考慮 audience 。如需詳細資訊，請參閱使用動態應用程式註冊的 Auth0 社群 - JWT 權杖
建立名為 mcp_auth0_client.py 的檔案，並貼上以下程式碼。此用戶端會處理 Auth0-specific要求，包括對象參數：
注意
此程式碼包含 httpx 修補，可將 User-Agent 標頭注入所有 HTTP 請求。這是必要的，因為 MCP Python SDK 目前在其 HTTP 請求中不包含使用者代理程式標頭，這可能會導致 AWS WAF 規則需要使用者代理程式標頭的問題。如需詳細資訊，請參閱 MCP Python SDK 問題 #1664
#!/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()
若要使用 用戶端：
-
設定必要的環境變數：
export AGENT_ARN="arn:aws:bedrock:us-west-2:123456789012:agent/ABCD1234" -
設定 Auth0-specific環境變數 （僅適用於 Auth0)：
export AUTH0_API_IDENTIFIER="your-api-identifier" -
執行用戶端：
python mcp_auth0_client.py
用戶端會自動：
-
編碼要在 URL 中使用的代理程式 ARN
-
建構 MCP 調用端點 URL
-
將 Auth0
audience參數新增至授權請求 （使用 Auth0 時） -
使用任何符合 OAuth 2.0 規範的身分提供者
附錄
設定 Cognito 使用者集區以進行身分驗證
建立新的檔案setup_cognito.sh並新增下列內容。
#!/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"
開啟終端機視窗並設定下列環境變數：
-
REGION– 您要使用的 AWS 區域 -
USERNAME– 新使用者的使用者名稱 -
PASSWORD– 新使用者的密碼
export REGION=us-east-1 // set your desired Region export USERNAME=USER NAME export PASSWORD=PASSWORD
使用命令 執行指令碼source setup_cognito.sh。
注意
如需詳細的 OAuth 身分驗證設定和服務連結角色資訊，請參閱使用傳入身分驗證和傳出身分驗證進行身分驗證和授權。
執行此指令碼後，請注意下列值以用於部署組態：
-
探索 URL：在
agentcore create步驟期間使用 -
用戶端 ID：在
agentcore create步驟期間使用 -
承載字符：調用部署的伺服器時使用
使用 MCP 檢查器進行本機測試
MCP Inspector 是一種視覺化工具，用於測試 MCP 伺服器。若要使用它，您需要：
-
已安裝 Node.js 和 npm
安裝並執行 MCP Inspector：
npx @modelcontextprotocol/inspector
這將：
-
啟動 MCP Inspector 伺服器
-
在終端機中顯示 URL （通常是
http://localhost:6274)
若要使用 Inspector：
-
在瀏覽器
http://localhost:6274中導覽至 -
將 MCP 伺服器 URL
http://localhost:8000/mcp() 貼到 MCP Inspector 連線欄位中 -
您會在側邊列看到您的工具
-
按一下任何工具進行測試
-
填寫參數 （例如，針對
add_numbers，輸入a和b的值） -
按一下「呼叫工具」以查看結果
使用 MCP 檢查器進行遠端測試
您也可以使用 MCP Inspector 測試已部署的伺服器。首先，對代理程式 ARN 進行 URL 編碼：
export AGENT_ARN="arn:aws:bedrock-agentcore:us-west-2:123456789012:runtime/my_mcp_server-xyz123" echo -n $AGENT_ARN | jq -sRr '@uri'
這會輸出 URL 編碼的 ARN：
arn%3Aaws%3Abedrock-agentcore%3Aus-west-2%3A123456789012%3Aruntime%2Fmy_mcp_server-xyz123
然後與 MCP Inspector 連線：
-
啟動 MCP Inspector：
npx @modelcontextprotocol/inspector -
在 Web 界面中：
-
選取「可串流 HTTP」做為傳輸
-
使用編碼的 ARN 輸入代理程式的端點 URL。請務必使用與客服人員 ARN 相同的區域：
https://bedrock-agentcore.REGION.amazonaws.com/runtimes/ENCODED_ARN/invocations?qualifier=DEFAULTus-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 -
在身分驗證區段中新增具有標頭名稱
Authorization和值的承載字符Bearer YOUR_TOKEN -
按一下「連線」
-
-
像在本機一樣測試您的工具
