本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。
在 AgentCore 运行时部署 MCP 服务器
Amazon Bedrock R AgentCore untime 允许您在运行时部署和运行模型上下文协议 (MCP) 服务器。 AgentCore 本指南引导您创建、测试和部署您的第一台 MCP 服务器。
有关示例，请参阅上的 AgentCore MCP 服务器基础知识。 GitHub
在本部分中，您将学习：
-
如何使用工具创建 MCP 服务器
-
如何在本地测试服务器
-
如何将服务器部署到 AWS
-
如何调用已部署的服务器
有关 MCP 的更多信息，请参阅 MCP 协议合同。
主题
亚马逊 Bedrock 如何 AgentCore 支持 MCP
当您使用 MCP 协议配置 Amazon Bedrock AgentCore Runtime 时，该服务期望 MCP 服务器容器在该路径上可用0.0.0.0:8000/mcp，这是大多数官方 MCP 服务器软件开发工具包支持的默认路径。
亚马逊 Bedrock AgentCore 支持无状态和有状态的 Streamable-HTTP MCP 服务器。默认情况下，建议基本 MCP 服务器使用无状态模式 (stateless_http=True)。该平台会自动为任何没有Mcp-Session-Id标头的请求添加标头，因此 MCP 客户端可以保持与同一 Amazon Bedrock AgentCore 运行时会话的连接连续性。
对于需要多回合交互（引发）、 LLM-generated 内容（采样）或进度通知的 MCP 服务器，状态模式（stateless_http=False）启用这些功能。在有状态模式下，运行时会保留同一次调用中请求之间的 MCP 会话状态。有关更多信息，请参阅有状态 MCP 服务器功能。
InvokeAgentRuntimeAPI 的有效载荷直接传递，从而可以轻松代理 MCP 等协议的 RPC 消息。
先决条件
在部署 MCP 服务器之前，请确保您具备以下条件：
-
已安装 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-generated 内容（采样）的 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 检查器测试服务器，如使用 MCP 检查器进行本地测试中所述。
第 3 步：将 MCP 服务器部署到 AWS
安装部署工具
安装 C AgentCore LI：
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 Runtime 所需的要求。
创建要部署的项目
在创建项目之前，您需要按照设置 Cognito 用户池进行身份验证中所述设置用于身份验证的 Cognito 用户池。这提供了安全访问已部署服务器所需的 OAuth 令牌。
注意
从 2025 年 10 月 7 日起，Amazon Bedrock 在 AgentCore 使用 OAuth 身份验证时使用 Service-Linked 角色来获得工作负载身份权限。有关此更改的详细信息，请参阅身份服务相关角色。
设置身份验证后，使用 MCP 协议构建一个新项目：
agentcore create --project-name MCPServerProject --no-agent cd MCPServerProject agentcore add agent \ --name MCPServer \ --language Python \ --protocol MCP \ --authorizer-type CUSTOM_JWT \ --discovery-url "https://cognito-idp.$REGION.amazonaws.com/$POOL_ID/.well-known/openid-configuration" \ --allowed-clients "$CLIENT_ID" \ --request-header-allowlist Authorization cp ../my_mcp_server.py app/MCPServer/main.py cd app/MCPServer uv add mcp cd ../..
CLI 创建 CUSTOM_JWT 运行时配置并构建项目结构。这些命令将您的服务器复制到生成的app/MCPServer/main.py入口点上，并将其依赖项添加到。pyproject.toml
部署到 AWS
部署您的代理：
agentcore deploy
此命令将：
-
打包您的代理代码和依赖关系
-
将部署项目上传到 Amazon S3
-
创建亚马逊 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标题，则它必须遵循 MCP application/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 检查器测试已部署的服务器，如使用 MCP 检查器进行远程测试中所述。
OAuth-Configured 代理的身份验证错误响应
OAuth-configured 代理遵循 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，因为它支持动态客户端注册 (DCR)，它允许客户端在运行时以编程方式注册自己，从而简化了客户端设置过程。
第 1 步-第 3 步：创建并测试您的 MCP 服务器
按照步骤 1 中的步骤 1-3：通过步骤 3：将 MCP 服务器部署 AWS到来创建和测试 MCP 服务器。
步骤 4：创建 Auth0 应用程序
按照 Okta 的 Auth0 上的 Auth0 设置说明进行操作。
启用动态客户机注册：
-
控制面板 → 设置 → 高级
-
切换 “OIDC 动态应用程序注册” → 开启
-
保存更改
有关更多信息，请参阅 Auth0 动态客户端注册文档
第 5 步：创建要部署的项目
设置身份验证后，使用 MCP 协议构建一个新项目：
agentcore create --project-name MCPServerProject --no-agent cd MCPServerProject agentcore add agent \ --name MCPServer \ --language Python \ --protocol MCP \ --authorizer-type CUSTOM_JWT \ --discovery-url "<AUTH0_DISCOVERY_URL>" \ --allowed-clients "<AUTH0_CLIENT_ID>" \ --request-header-allowlist Authorization cp ../my_mcp_server.py app/MCPServer/main.py cd app/MCPServer uv add mcp cd ../..
将 Auth0 占位符替换为 Auth0 应用程序中的值。CLI 创建 CUSTOM_JWT 运行时配置并构建项目结构。这些命令将您的服务器复制到生成的app/MCPServer/main.py入口点上，并将其依赖项添加到。pyproject.toml
步骤 6：部署到 AWS
部署您的代理：
agentcore deploy
此命令将：
-
打包您的代理代码和依赖关系
-
将部署项目上传到 Amazon S3
-
创建亚马逊 Bedrock 运行 AgentCore 时
-
将您的代理部署到 AWS
部署后，您将收到一个代理运行时 ARN，如下所示：
arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_mcp_server-xyz123
步骤 7：调用已部署的代理
该客户端基于官方的 MCP SDK 简单身份验证客户端示例，并进行了修改。
注意
在动态客户端注册中使用 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 补丁，用于在所有 HTTP 请求中注入 User-Agent 标头。这是必要的，因为 MCP Python SDK 目前在其 HTTP 请求中不包含 User-Agent 标头，这可能会导致需要 User-Agent 标头的 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
客户端将自动：
-
对代理 ARN 进行编码以在 URL 中使用
-
构造 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 身份验证设置和 Service-Linked 角色信息，请参阅使用入站身份验证和出站身份验证进行身份验证和授权。
运行此脚本后，记下用于部署配置的以下值：
-
发现 URL：在
agentcore create步骤中使用 -
客户端 ID：在
agentcore create步骤中使用 -
持有者令牌：在调用已部署的服务器时使用
使用 MCP 检查器进行本地测试
MCP 检查器是用于测试 MCP 服务器的可视化工具。要使用它，你需要：
-
Node.js 并安装了 npm
安装并运行 MCP 检查器：
npx @modelcontextprotocol/inspector
这将：
-
启动 MCP Inspector 服务器
-
在终端中显示 URL（通常
http://localhost:6274）
要使用检查器，请执行以下操作：
-
http://localhost:6274在浏览器中导航至 -
将 MCP 服务器 URL (
http://localhost:8000/mcp) 粘贴到 MCP 检查器连接字段中 -
你将在侧栏中看到你的工具
-
点击任何工具进行测试
-
填写参数（例如，对于
add_numbers，输入a和的值b） -
点击 “呼叫工具” 查看结果
使用 MCP 检查器进行远程测试
您还可以使用 MCP 检查器测试已部署的服务器。首先， URL-encode 您的代理 ARN：
export AGENT_ARN="arn:aws:bedrock-agentcore:us-west-2:123456789012:runtime/my_mcp_server-xyz123" echo -n $AGENT_ARN | jq -sRr '@uri'
这将输出 URL-encoded ARN：
arn%3Aaws%3Abedrock-agentcore%3Aus-west-2%3A123456789012%3Aruntime%2Fmy_mcp_server-xyz123
然后与 MCP 检查器连接：
-
启动 MCP 检查器：
npx @modelcontextprotocol/inspector -
在 Web 界面中：
-
选择 “可流式传输的 HTTP” 作为传输
-
使用编码的 ARN 输入代理的终端节点 URL。确保使用与代理的 ARN 相同的区域：
https://bedrock-agentcore.REGION.amazonaws.com/runtimes/ENCODED_ARN/invocations?qualifier=DEFAULT以 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 -
在身份验证部分添加带有标头名称
Authorization和值的持有者令牌Bearer YOUR_TOKEN -
点击 “连接”
-
-
像在本地一样测试工具
