# 在运行时部署 MCP 服务器 AgentCore
Amazon Bedrock R AgentCore untime 允许您在运行时中部署和运行模型上下文协议 (MCP) 服务器。 AgentCore 本指南将引导您创建、测试和部署您的第一台 MCP 服务器。
有关示例，请参阅[https://github.com/awslabs/amazon-bedrock-agentcore-samples/tree/main/01-tutorials/01-AgentCore-runtime/02-hosting-MCP-server](https://github.com/awslabs/amazon-bedrock-agentcore-samples/tree/main/01-tutorials/01-AgentCore-runtime/02-hosting-MCP-server)。
在本部分中，您将学习：
+ 如何使用工具创建 MCP 服务器
+ 如何在本地测试服务器
+ 如何将服务器部署到 AWS
+ 如何调用已部署的服务器
有关 MCP 的更多信息，请参阅 [MCP 协议](runtime-mcp-protocol-contract.md)合约。
**Topics**
+ [Amazon Bedrock 如何 AgentCore 支持 MCP](#runtime-mcp-how-it-works)
+ [先决条件](#runtime-mcp-prerequisites)
+ [步骤 1：创建您的 MCP 服务器](#runtime-mcp-create-server)
+ [第 2 步：在本地测试您的 MCP 服务器](#runtime-mcp-test-locally)
+ [步骤 3：将 MCP 服务器部署到 AWS](#runtime-mcp-deploy-aws)
+ [步骤 4：调用已部署的 MCP 服务器](#runtime-mcp-invoke-server)
+ [OAuth-Configured 代理的身份验证错误响应](#runtime-mcp-auth-error-responses)
+ [使用 Auth0 进行端到端流程](#runtime-mcp-auth0-flow)
+ [附录](#runtime-mcp-appendix)
## 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 服务器功能](mcp-stateful-features.md)。
[InvokeAgentRuntime](https://docs.aws.amazon.com/bedrock-agentcore/latest/APIReference/API_InvokeAgentRuntime.html)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 工具**
+ **工具**：演示不同类型操作的三种简单工具
+ stat@@ **eless\_http=True**：将服务器配置为无状态模式，这是基本 MCP 服务器的默认模式
**提示**
对于需要多回合交互（激发）或 LLM-generated 内容（采样）的 MCP 服务器，`stateless_http=False`请使用启用状态模式。有状态 MCP 服务器在同一个工具调用中维护多个请求的会话上下文。有关更多信息，请参阅[有状态 MCP 服务器功能](mcp-stateful-features.md)。
## 第 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 检查器[进行本地测试](#runtime-mcp-appendix-b)中所述。
## 步骤 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 用户](#runtime-mcp-appendix-a)池进行身份验证中所述。这提供了安全访问已部署服务器所需的 OAuth 令牌。
**注意**
从 **2025 年 10 月 7 日**起，Amazon Bedrock 在 AgentCore 使用 OAuth 身份验证时使用 Service-Linked 角色来获得工作负载身份权限。有关此更改的详细信息，请参阅[身份服务相关角色](service-linked-roles.md#identity-service-linked-role)。
设置身份验证后，使用 MCP 协议搭建一个新项目：
```
agentcore create --protocol MCP
```
按照交互式提示提供项目名称。CLI 为包括`agentcore/agentcore.json`配置文件在内的项目结构奠定了基础。将您的`my_mcp_server.py`文件复制到生成的项目的代理代码目录中，并确保入口点指向您的服务器文件。`agentcore/agentcore.json`
### 部署到 AWS
部署您的代理：
```
agentcore deploy
```
此命令将：
1. Package 你的代理代码和依赖关系
1. 将部署项目上传到 Amazon S3
1. 创建 Amazon Bedrock 运行 AgentCore 时
1. 将您的代理部署到 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](https://modelcontextprotocol.io/specification/2025-06-18/basic/transports#sending-messages-to-the-server) 标准。可接受的媒体类型为`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 Inspector 测试已部署的服务器，如使用 MCP 检查[器进行远程测试](#runtime-mcp-appendix-c)中所述。
## OAuth-Configured 代理的身份验证错误响应
OAuth-configured 代理遵守 [RFC 6749 (OAuth 2.0](https://datatracker.ietf.org/doc/html/rfc6749)) 身份验证标准。当缺少身份验证时，该服务会返回带有 WWW-Authenticate 标头的 401 未经授权的响应（根据 [RFC 7235](https://datatracker.ietf.org/doc/html/rfc7235)），使客户端能够通过 API 发现授权服务器端点。 GetRuntimeProtectedResourceMetadata
### 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 服务器中的](#runtime-mcp-create-server)步骤 1-[3 进行操作：将 MCP 服务器部署 AWS到以](#runtime-mcp-deploy-aws)创建和测试 MCP 服务器。
### 步骤 4：创建 Auth0 应用程序
按照 Okta 的 Auth0 上的 [Auth0 设置说明进行](identity-idp-auth0.md)操作。
**启用动态客户端注册：**
1. 控制面板 → 设置 → 高级
1. 切换 “OIDC 动态应用程序注册” → “开”
1. 保存更改
有关更多信息，请参阅 [Auth0 动态客户端注册文档](https://auth0.com/docs/get-started/applications/dynamic-client-registration)。
### 步骤 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](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/clients/simple-auth-client/mcp_simple_auth_client/main.py) lient 示例，并进行了修改。 Auth0-specific
**注意**
在动态客户端注册中使用 Auth0 时，必须在授权请求中包含`audience`参数才能接收 JWT 令牌。如果没有此参数，Auth0 将返回不透明令牌或 JWE（加密）令牌，而不是标准 JWT 令牌。MCP SDK 发送 OAuth 2.0 的`resource`参数 (RFC 8707)，但是 Auth0 需要 JWT 令牌使用 OIDC 参数。`audience`这两个参数的用途相似，但 Auth0 会优先考虑`audience`。有关更多信息，请参阅 [Auth0 社区-使用动态应用程序注册的 JWT 令牌](https://community.auth0.com/t/jwt-tokens-with-dynamic-application-registration/189741)。
使用以下代码创建名为 `mcp_auth0_client.py` 的文件。此客户端处理包括受众参数在内的 Auth0-specific 需求：
**注意**
该代码包括 httpx 修补，用于向所有 HTTP User-Agent 请求注入标头。这是必要的，因为 MCP Python SDK 目前不在其 HTTP 请求中包含 User-Agent 标头，这可能会导致需要 User-Agent 标头的 AWS WAF 规则出现问题。有关更多信息，请参阅 [MCP Python SDK 问题 \#1664](https://github.com/modelcontextprotocol/python-sdk/issues/1664) 和 [AWS WAF 托管规则](https://docs.aws.amazon.com/waf/latest/developerguide/aws-managed-rule-groups-baseline.html)组。
#### MCP Python Auth0 客户端代码
```
#!/usr/bin/env python3
"""
MCP client with OAuth authentication support for Auth0.
Based on the official MCP SDK simple-auth-client example with Auth0 compatibility.
Adds support for Auth0's 'audience' parameter requirement.
Usage:
# Required
export AGENT_ARN="arn:aws:bedrock:us-west-2:123456789012:agent/ABCD1234"
# Required for Auth0
export AUTH0_API_IDENTIFIER="your-api-identifier"
# Optional - custom endpoint for beta/dev environments
export CUSTOM_ENDPOINT="https://beta.example.com"
python mcp_auth0_client.py
The client will automatically:
- Encode the Agent ARN for use in the URL
- Construct the MCP invocation endpoint URL
- Add Auth0 'audience' parameter to authorization requests (when using Auth0)
- Work with any OAuth 2.0 compliant identity provider
"""
import asyncio
import httpx
import os
import threading
import time
import webbrowser
from datetime import timedelta
from http.server import BaseHTTPRequestHandler, HTTPServer
from typing import Any
from urllib.parse import parse_qs, urlencode, urlparse, urlunparse
# Patch httpx at the request level to inject User-Agent header
# This ensures ALL HTTP requests have the User-Agent header, including OAuth discovery calls
_original_httpx_request = httpx.Request.__init__
def _patched_httpx_request_init(self, method, url, *args, **kwargs):
"""Patched Request.__init__ that injects User-Agent header into all HTTP requests."""
# Get or create headers
headers = kwargs.get('headers')
if headers is None:
headers = {}
kwargs['headers'] = headers
# Convert to mutable dict if needed
if not isinstance(headers, dict):
headers = dict(headers)
kwargs['headers'] = headers
# Inject User-Agent if not present (case-insensitive check)
if 'User-Agent' not in headers and 'user-agent' not in headers:
headers['User-Agent'] = 'python-mcp-sdk/1.0 (BedrockAgentCore-Runtime)'
# Call original __init__
_original_httpx_request(self, method, url, *args, **kwargs)
# Apply the patch globally before importing MCP modules
httpx.Request.__init__ = _patched_httpx_request_init
# Now import MCP modules - they will use patched httpx
from mcp.client.auth import OAuthClientProvider, TokenStorage
from mcp.client.session import ClientSession
from mcp.client.sse import sse_client
from mcp.client.streamable_http import streamablehttp_client
from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken
class InMemoryTokenStorage(TokenStorage):
"""Simple in-memory token storage implementation."""
def __init__(self):
self._tokens: OAuthToken | None = None
self._client_info: OAuthClientInformationFull | None = None
async def get_tokens(self) -> OAuthToken | None:
return self._tokens
async def set_tokens(self, tokens: OAuthToken) -> None:
self._tokens = tokens
async def get_client_info(self) -> OAuthClientInformationFull | None:
return self._client_info
async def set_client_info(self, client_info: OAuthClientInformationFull) -> None:
self._client_info = client_info
class CallbackHandler(BaseHTTPRequestHandler):
"""Simple HTTP handler to capture OAuth callback."""
def __init__(self, request, client_address, server, callback_data):
"""Initialize with callback data storage."""
self.callback_data = callback_data
super().__init__(request, client_address, server)
def do_GET(self):
"""Handle GET request from OAuth redirect."""
parsed = urlparse(self.path)
query_params = parse_qs(parsed.query)
if "code" in query_params:
self.callback_data["authorization_code"] = query_params["code"][0]
self.callback_data["state"] = query_params.get("state", [None])[0]
self.send_response(200)
self.send_header("Content-type", "text/html")
self.end_headers()
self.wfile.write(b"""
Authorization Successful!
You can close this window and return to the terminal.
""")
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"""
Authorization Failed
Error: {query_params["error"][0]}
You can close this window and return to the terminal.
""".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 [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 ', 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()
```
要使用客户端，请执行以下操作：
1. 设置所需的环境变量：
```
export AGENT_ARN="arn:aws:bedrock:us-west-2:123456789012:agent/ABCD1234"
```
1. 设置 Auth0-specific 环境变量（仅适用于 Auth0）：
```
export AUTH0_API_IDENTIFIER="your-api-identifier"
```
1. 运行客户端：
```
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 角色的详细信息，请参阅[使用入站身份验证和出站身份验证进行身份验证和授权](runtime-oauth.md)。
运行此脚本后，请记下以下值以便在部署配置中使用：
+ 发现 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：
1. `http://localhost:6274`在浏览器中导航至
1. 将 MCP 服务器 URL (`http://localhost:8000/mcp`) 粘贴到 MCP Inspector 连接字段中
1. 你将在侧栏中看到你的工具列出
1. 点击任何工具进行测试
1. 填写参数（例如，对于`add_numbers`，输入`a`和的值`b`）
1. 点击 “调用工具” 查看结果
### 使用 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：
1. 启动 MCP Inspector：
```
npx @modelcontextprotocol/inspector
```
1. 在 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`
+ 点击 “Connect”
1. 像在本地一样测试您的工具