在运行时部署 MCP 服务器 AgentCore
Amazon Bedrock R AgentCore untime 允许您在运行时中部署和运行模型上下文协议 (MCP) 服务器。 AgentCore 本指南将引导您创建、测试和部署您的第一台 MCP 服务器。
在本部分中，您将学习：
-
如何使用工具创建 MCP 服务器
-
如何在本地测试服务器
-
如何将服务器部署到 AWS
-
如何调用已部署的服务器
有关 MCP 的更多信息，请参阅 MCP 协议合约。
主题
Amazon Bedrock 如何 AgentCore 支持 MCP
当您使用 MCP 协议配置 Amazon Bedrock R AgentCore untime 时，该服务希望 MCP 服务器容器在该路径上可用0.0.0.0:8000/mcp，这是大多数官方 MCP 服务器软件开发工具包支持的默认路径。
Amazon 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 消息。
先决条件
-
已安装 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 工具
-
工具：演示不同类型操作的三种简单工具
-
stat@@ eless_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 Inspector 测试您的服务器，如使用 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 运行时所需的要求。
创建要部署的项目
在创建项目之前，您需要设置 Cognito 用户池进行身份验证，如设置 Cognito 用户池进行身份验证中所述。这提供了安全访问已部署服务器所需的 OAuth 令牌。
注意
从 2025 年 10 月 7 日起，Amazon Bedrock 在 AgentCore 使用 OAuth 身份验证时使用 Service-Linked 角色来获得工作负载身份权限。有关此更改的详细信息，请参阅身份服务相关角色。
设置身份验证后，使用 MCP 协议搭建一个新项目：
agentcore create --protocol MCP
按照交互式提示提供项目名称。CLI 为包括agentcore/agentcore.json配置文件在内的项目结构奠定了基础。将您的my_mcp_server.py文件复制到生成的项目的代理代码目录中，并确保入口点指向您的服务器文件。agentcore/agentcore.json
部署到 AWS
部署您的代理：
agentcore deploy
此命令将：
-
Package 你的代理代码和依赖关系
-
将部署项目上传到 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-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：创建 MCP 服务器中的步骤 1-3 进行操作：将 MCP 服务器部署 AWS到以创建和测试 MCP 服务器。
步骤 4：创建 Auth0 应用程序
按照 Okta 的 Auth0 上的 Auth0 设置说明进行操作。
启用动态客户端注册：
-
控制面板 → 设置 → 高级
-
切换 “OIDC 动态应用程序注册” → “开”
-
保存更改
有关更多信息，请参阅 Auth0 动态客户端注册文档
步骤 5：创建要部署的项目
设置身份验证后，使用 MCP 协议搭建一个新项目：
agentcore create --protocol MCP
按照交互式提示提供项目名称。CLI 为包括agentcore/agentcore.json配置文件在内的项目结构奠定了基础。将您的my_mcp_server.py文件复制到生成的项目的代理代码目录中，并确保入口点指向您的服务器文件。agentcore/agentcore.json
步骤 6：部署到 AWS
部署您的代理：
agentcore deploy
此命令将：
-
Package 你的代理代码和依赖关系
-
将部署项目上传到 Amazon S3
-
创建 Amazon Bedrock 运行 AgentCore 时
-
将您的代理部署到 AWS
部署后，您将收到一个代理运行时 ARN，如下所示：
arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_mcp_server-xyz123
步骤 7：调用已部署的代理
该客户端基于官方 MCP SDK simple-auth-c
注意
在动态客户端注册中使用 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 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 测试已部署的服务器。首先， 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 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 -
点击 “Connect”
-
-
像在本地一样测试您的工具
