📢 转载信息
原文作者：Frank Dallezotte, Dhawalkumar Patel, and Ganesh Thiyagarajan
随着AI代理被大规模采用，开发团队可以创建数十到数百个专业化的模型上下文协议（MCP）服务器，这些服务器针对特定的代理用例、领域、组织功能或团队进行了定制。组织还需要集成自己现有的MCP服务器或开源MCP服务器到其AI工作流程中。因此，需要一种有效的方法将这些现有的MCP服务器——无论是定制构建的、公开可用的还是开源的——组合成一个统一的接口，供AI代理轻松消费，并供团队在组织内部无缝共享。
今年早些时候，我们推出了Amazon Bedrock AgentCore Gateway，这是一个完全托管的服务，充当集中的MCP工具服务器，提供了一个统一的接口，代理可以在其中发现、访问和调用工具。今天，我们正在扩展对现有MCP服务器的支持，将其作为AgentCore Gateway中的一种新目标类型。通过此功能，您可以将多个与代理目标对齐的任务特定MCP服务器分组到一个单一的、可管理的MCP网关接口之下。这减少了维护单独网关的操作复杂性，同时提供了与REST API和AWS Lambda函数已有的集中式工具和身份验证管理功能。
如果没有集中化的方法，客户将面临重大挑战：跨组织发现和共享工具变得分散，管理多个MCP服务器的身份验证日益复杂，为每个服务器维护单独的网关实例很快就会变得难以管理。Amazon Bedrock AgentCore Gateway通过将现有MCP服务器视为原生目标来帮助解决这些挑战，为路由、身份验证和工具管理提供了一个单一的控制点——集成MCP服务器就像向网关添加其他目标一样简单。
打破MCP孤岛：为什么企业团队需要一个统一的网关
让我们通过一个电子商务订单系统的真实示例来探讨这一点，在这个系统中，不同的团队为其特定领域维护专业的MCP服务器。考虑一个企业电子商务系统，其中不同的团队为其特定领域开发了专业的MCP服务器：
- 购物车团队维护一个带有购物车管理工具的MCP服务器
- 产品目录团队运行其MCP服务器用于产品浏览和搜索
- 促销团队操作一个处理促销逻辑的MCP服务器
以前，一个订单代理需要单独与这些MCP服务器交互，管理多个连接和身份验证上下文。借助AgentCore Gateway中新的MCP服务器目标支持，这些专业化的服务器现在可以统一在一个网关下，同时保持其团队特定的所有权和访问控制。这种方法的强大之处在于其组织灵活性。团队可以根据多种逻辑标准对MCP服务器进行分组：
- 业务单元对齐：按业务单元组织MCP服务器
- 产品功能边界：每个产品团队拥有其带有领域特定工具的MCP服务器，允许他们保持清晰的所有权，同时为其代理提供统一的接口
- 安全和访问控制：不同的MCP服务器需要不同的身份验证机制。网关处理身份验证的复杂性，使授权代理能够轻松访问他们需要的工具
下图说明了订购代理如何通过AgentCore Gateway与多个MCP服务器交互。代理连接到网关并发现可用工具。每个团队都可以控制其领域特定的工具，同时为统一的代理体验做出贡献。网关处理工具命名冲突、身份验证，并跨工具提供统一的语义搜索。
AgentCore Gateway在现代代理架构中充当集成中心，为连接各种代理实现与各种工具提供商提供统一的接口。如图所示的架构，演示了网关如何弥合代理和工具实现方法之间的差距，现在通过直接集成MCP服务器目标的能力得到了增强。
AgentCore Gateway集成架构
在AgentCore Gateway中，目标定义了网关将作为工具提供给代理的API、Lambda函数或其他MCP服务器。目标可以是Lambda函数、OpenAPI规范、Smithy模型、MCP服务器或其他工具定义。
架构的目标集成方面展示了网关在工具集成方面的多功能性。借助新的MCP服务器目标支持，网关可以直接整合来自公共MCP服务器的工具，将它们与其他目标类型一起视为一流公民。此功能扩展到联合场景，其中一个AgentCore Gateway实例可以作为另一个实例的目标，用于跨组织边界的层次化工具组织。网关可以无缝集成使用代理作为工具暴露的AgentCore Runtime实例、客户维护的私有MCP服务器、传统的AWS Lambda函数以及Smithy和AWS服务API。
除了目标多样性之外，网关的身份验证架构还提供了额外的操作优势。网关将其入站身份验证与其目标系统解耦，允许代理通过单个接口访问使用多个身份提供商的工具。这种集中化的方法简化了AI代理的开发、部署和维护。现在，对于MCP服务器目标，可以使用相同的方法，其中网关使用为目标配置的身份提供商来管理与服务器交互的复杂性。
借助此身份验证基础，您可以通过统一的架构获得复杂的工具管理功能。当代理请求工具发现时，网关在集成目标中提供一致的视图，来自MCP服务器的工具与Lambda函数和传统API并列出现。语义搜索功能在工具类型上统一运行，因此代理可以发现相关的工具，而无论其实现如何。在工具调用期间，网关处理必要的协议转换、身份验证流程和数据转换，在幕后管理不同目标系统的复杂性，同时向代理呈现干净、一致的接口。
添加MCP服务器目标支持代表了网关功能的重大演进。组织现在可以直接集成MCP原生的工具，同时维护对传统API和Lambda函数的投资。这种灵活性允许采用渐进式的迁移策略，团队可以按照自己的进度采用MCP原生的实现，同时促进现有集成的持续运行。网关的同步机制确保工具定义在不同目标类型中保持最新，而其身份验证和授权系统无论底层工具实现如何，都能提供一致的安全控制。
该网关将MCP服务器、传统API和无服务器函数组合成一个连贯的工具环境。此功能以及企业级的安全性和性能，使其成为代理计算的有利基础设施。
解决方案演练
在本文中，我们将指导您完成在AgentCore Gateway中设置MCP服务器目标的步骤，这与将新的MCP服务器类型目标添加到新的或现有的MCP网关一样简单。将MCP服务器添加到AgentCore Gateway将允许您在规模化管理MCP服务器时，集中您的工具管理、安全身份验证和操作最佳实践。
开始将MCP服务器添加到AgentCore Gateway
要开始，您将创建一个AgentCore Gateway并将您的MCP服务器添加为目标。
先决条件
验证您具有以下先决条件：
- 具有Amazon Bedrock AgentCore访问权限的AWS账户。有关更多信息，请参阅AgentCore Runtime权限文档。
- Python 3.12或更高版本
- 对OAuth 2.0的基本了解
您可以通过多种界面创建网关并添加目标：
以下实际示例和代码片段演示了如何设置和使用Amazon Bedrock AgentCore Gateway。有关交互式演练，您可以使用这些GitHub上的Jupyter Notebook示例。
创建网关
要创建网关，您可以使用AgentCore入门工具包，使用Amazon Cognito为基于JWT的入站身份验证创建默认授权配置。您也可以使用另一个OAuth 2.0兼容的身份提供商代替Cognito。
import time
import boto3
gateway_client = boto3.client("bedrock-agentcore-control")
# Create an authorization configuration, that specifies what client is authorized to access this Gateway
auth_config = { "customJWTAuthorizer": { "allowedClients": ['<cognito_client_id>'], # Client MUST match with the ClientId configured in Cognito. "discoveryUrl": '<cognito_oauth_discovery_url>', }
}
# Call the create_gateway API
# This operation is asynchronous so may take time for Gateway creation
# This Gateway will leverage a CUSTOM_JWT authorizer, the Cognito User Pool we reference in auth_config
def deploy_gateway(poll_interval=5):
create_response = gateway_client.create_gateway(
name="DemoGateway",
roleArn="<IAM Role>", # The IAM Role must have permissions to create/list/get/delete Gateway
protocolType="MCP",
authorizerType="CUSTOM_JWT",
authorizerConfiguration=auth_config,
description="AgentCore Gateway with MCP Server Target",
)
gatewayID = create_response["gatewayId"]
gatewayURL = create_response["gatewayUrl"]
# Wait for deployment
while True:
status_response = gateway_client.get_gateway(gatewayIdentifier=gatewayID)
status = status_response["status"]
if status == "READY":
print("✅ AgentCore Gateway is READY!")
break
elif status in ["FAILED"]:
print(f"❌ Deployment failed: {status}")
return None
print(f"Status: {status} - waiting...")
time.sleep(poll_interval)
if __name__ == "__main__":
deploy_gateway()
# Values with < > needs to be replaced with real values创建一个示例MCP服务器
作为一个例子，让我们创建一个带有三个返回静态响应的简单工具的示例MCP服务器。该服务器使用FastMCP，并设置stateless_http=True，这是AgentCore Runtime兼容性所必需的。
from mcp.server.fastmcp import FastMCP
mcp = FastMCP(host="0.0.0.0", stateless_http=True)
@mcp.tool()
def getOrder() -> int:
"""Get an order"""
return 123
@mcp.tool()
def updateOrder(orderId: int) -> int:
"""Update existing order"""
return 456
@mcp.tool()
def cancelOrder(orderId: int) -> int:
"""cancel existing order"""
return 789
if __name__ == "__main__":
mcp.run(transport="streamable-http")配置AgentCore Runtime部署
接下来，我们将使用入门工具包来配置AgentCore Runtime部署。该工具包可以在启动时创建Amazon ECR存储库，并为AgentCore Runtime上的部署生成Dockerfile。您可以使用自己现有的MCP服务器，我们在此仅用作示例。在实际环境中，您的MCP服务器的入站身份验证可能与网关配置不同。请参阅此GitHub代码示例以为Runtime身份验证创建Amazon Cognito用户池。
from bedrock_agentcore_starter_toolkit import Runtime
from boto3.session import Session
boto_session = Session()
region = boto_session.region_name
print(f"Using AWS region: {region}")
required_files = ['mcp_server.py', 'requirements.txt']
for file in required_files:
if not os.path.exists(file):
raise FileNotFoundError(f"Required file {file} not found")
print("All required files found ✓")
agentcore_runtime = Runtime()
auth_config = {
"customJWTAuthorizer": {
"allowedClients": [
'<runtime_cognito_client_id>' # Client MUST match with the ClientId configured in Cognito, and can be separate from the Gateway Cognito provider.
],
"discoveryUrl": '<cognito_oauth_discovery_url>',
}
}
print("Configuring AgentCore Runtime...")
response = agentcore_runtime.configure(
entrypoint="mcp_server.py",
auto_create_execution_role=True,
auto_create_ecr=True,
requirements_file="requirements.txt",
region=region,
authorizer_configuration=auth_config,
protocol="MCP",
agent_name="mcp_server_agentcore"
)
print("Configuration completed ✓")
# Values with < > needs to be replaced with real values启动MCP服务器到AgentCore Runtime
现在我们有了Dockerfile，让我们将MCP服务器启动到AgentCore Runtime：
print("Launching MCP server to AgentCore Runtime...")
print("This may take several minutes...")
launch_result = agentcore_runtime.launch()
agent_arn = launch_result.agent_arn
agent_id = launch_result.agent_id
print("Launch completed ✓")
encoded_arn = agent_arn.replace(':', '%3A').replace('/', '%2F')
mcp_url = f"https://bedrock-agentcore.{region}.amazonaws.com/runtimes/{encoded_arn}/invocations?qualifier=DEFAULT"
print(f"Agent ARN: {launch_result.agent_arn}")
print(f"Agent ID: {launch_result.agent_id}")为AgentCore Gateway创建MCP服务器作为目标
为AgentCore Gateway创建一个AgentCore身份资源凭证提供商，以便在向AgentCore Runtime中的MCP服务器代理进行出站身份验证时使用：
identity_client = boto3.client('bedrock-agentcore-control', region_name=region)
cognito_provider = identity_client.create_oauth2_credential_provider(
name="gateway-mcp-server-identity",
credentialProviderVendor="CustomOauth2",
oauth2ProviderConfigInput={
'customOauth2ProviderConfig': {
'oauthDiscovery': {
'discoveryUrl': '<cognito_oauth_discovery_url>',
},
'clientId': '<runtime_cognito_client_id>', # Client MUST match with the ClientId configured in Cognito for the Runtime authorizer
'clientSecret': '<cognito_client_secret>'
}
}
)
cognito_provider_arn = cognito_provider['credentialProviderArn']
print(cognito_provider_arn)
# Values with < > needs to be replaced with real values创建一个指向MCP服务器的网关目标：
gateway_client = boto3.client("bedrock-agentcore-control", region_name=region)
create_gateway_target_response = gateway_client.create_gateway_target(
name="mcp-server-target",
gatewayIdentifier=gatewayID,
targetConfiguration={"mcp": {"mcpServer": {"endpoint": mcp_url}}},
credentialProviderConfigurations=[
{
"credentialProviderType": "OAUTH",
"credentialProvider": {
"oauthCredentialProvider": {
"providerArn": cognito_provider_arn,
"scopes": ["<cognito_oauth_scopes>"],
}
},
},
],
)
# Asynchronously create gateway target
gatewayTargetID = create_gateway_target_response["targetId"]
# Values with < > needs to be replaced with real values创建网关目标后，实现一个轮询机制，使用get_gateway_target API调用来检查网关目标的状态：
import time
def poll_for_status(interval=5):
# Poll for READY status
while True:
gateway_target_response = gateway_client.get_gateway_target(gatewayIdentifier=gatewayID, targetId=gatewayTargetID)
status = gateway_target_response["status"]
if status == 'READY':
break
elif status in ['FAILED', 'UPDATE_UNSUCCESSFUL', 'SYNCHRONIZE_UNSUCCESSFUL']:
raise Exception(f"Gateway target failed with status: {status}")
time.sleep(interval)
poll_for_status()
使用Strands Agents框架测试网关
让我们使用Strands Agents集成来测试网关，以列出MCP服务器中的工具。您也可以使用使用不同代理框架构建的其他MCP兼容代理。
from strands import Agent
from mcp.client.streamable_http import streamablehttp_client
from strands.tools.mcp.mcp_client import MCPClient
def create_streamable_http_transport():
return streamablehttp_client(gatewayURL,headers={"
Authorization": f"Bearer {token}"})
client = MCPClient(create_streamable_http_transport)
with client:
# Call the listTools
tools = client.list_tools_sync()
# Create an Agent with the model and tools
agent = Agent(model=yourmodel,tools=tools)
## you can replace with any model you like
# Invoke the agent with the sample prompt. This will only invoke MCP listTools and retrieve the list of tools the LLM has access to. The below does not actually call any tool.
agent("Hi , can you list all tools available to you")
# Invoke the agent with sample prompt, invoke the tool and display the response
agent("Get the Order id")刷新AgentCore Gateway中MCP服务器的工具定义
SynchronizeGatewayTargets API是一项新的异步操作，支持从MCP服务器目标按需同步工具。MCP服务器托管代理可以发现和调用的工具。随着时间的推移，这些工具可能需要更新，或者在现有的MCP服务器目标中可能引入新工具。您可以通过SynchronizeGatewayTargets API与外部MCP服务器连接，该API执行协议握手并索引可用工具。此API为客户提供了明确控制何时刷新其工具定义的能力，在对其MCP服务器的工具配置进行更改后尤其有用。
当目标配置了OAuth身份验证时，API首先与AgentCore Identity服务交互，以从指定的凭证提供商检索必要的凭证。在与MCP服务器通信开始之前，会验证这些凭证的新鲜度和可用性。如果凭证检索失败或返回过期令牌，同步操作将立即失败并显示适当的错误详细信息，将目标转换到FAILED状态。对于未配置身份验证的目标，API直接进行工具同步。
工具处理工作流程始于对MCP服务器的初始化调用，以建立会话。成功初始化后，API对MCP服务器的tools/list功能进行分页调用，以每批100个工具进行处理，以优化性能和资源利用率。每个工具批次都会经过规范化处理，其中API会添加特定于目标的前缀，以帮助防止与其他目标的工具发生命名冲突。在处理过程中，工具定义会规范化以促进不同目标类型之间的一致性，同时保留原始MCP服务器定义的基本元数据。
当发生以下情况时，同步流程开始：
- 运维管理员启动
SynchronizeGatewayTargetsAPI，触发AgentCore Gateway刷新配置的MCP目标。 - 网关从AgentCore Identity获取OAuth令牌，以安全访问MCP目标。
- 然后，网关与MCP服务器建立安全会话以检索版本能力。
- 最后，网关对MCP服务器的
tools/list端点进行分页调用以检索工具定义，确保网关维护一个当前且准确的工具列表。
SynchronizeGatewayTargets API解决了在AgentCore Gateway中管理MCP目标的一个关键挑战：在优化系统性能和资源利用率的同时，保持可用工具的准确表示。以下是这种显式同步方法有价值的原因：
Schema一致性管理：如果没有显式同步，AgentCore Gateway需要在ListTools操作期间对MCP服务器进行实时调用（影响延迟和可靠性），或者有提供过时工具定义的风险。SynchronizeGatewayTargets API提供了一种受控机制，客户可以在战略时间点刷新其工具架构，例如在MCP服务器中部署新工具或更新现有工具后。这种方法确保网关中的工具定义准确反映目标MCP服务器的能力，而不会影响性能。
- 性能影响权衡：API在同步期间实现乐观锁定，以帮助防止可能导致不一致状态的并发修改。虽然这意味着多个同步请求可能需要重试（如果存在争用），但这种权衡是可以接受的，因为：
- 工具架构更改通常是不频繁的操作事件，而不是常规的运行时事件
- 同步的性能成本仅在明确请求时发生，而不是在常规工具调用期间发生
- 缓存的工具定义有助于在同步期间保持
ListTools操作的一致的高性能
调用同步网关API
使用以下示例调用同步网关操作：
import requests
import json
def search_tools(gateway_url, access_token, query):
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {access_token}"
}
payload = {
"jsonrpc": "2.0",
"id": "search-tools-request",
"method": "tools/call",
"params": {
"name": "x_amz_bedrock_agentcore_search",
"arguments": {
"query": query
}
}
}
response = requests.post(gateway_url, headers=headers, json=payload, timeout=5)
response.raise_for_status()
return response.json()
# Example usage
token_response = utils.get_token(user_pool_id, client_id, client_secret, scopeString, REGION)
access_token = token_response['access_token']
results = search_tools(gatewayURL, access_token, "order operations")
print(json.dumps(results, indent=2))
工具架构的隐式同步
在CreateGatewayTarget和UpdateGatewayTarget操作期间，AgentCore Gateway执行的隐式同步与显式的SynchronizeGatewayTargets API不同。这种隐式同步确保使用有效、最新的工具定义创建或更新MCP目标，与AgentCore Gateway确保处于READY状态的目标可立即使用这一保证相一致。虽然这可能使创建/更新操作比使用其他目标类型耗时更长，但它有助于避免目标没有经过验证的工具定义所带来的复杂性和潜在问题。
当发生以下情况时，隐式同步流程开始：
- 运维管理员使用
CreateGatewayTarget或UpdateGatewayTarget操作创建或更新MCP目标。 - AgentCore Gateway配置新创建或更新的MCP目标。
- 网关异步触发同步过程以更新工具定义。
- 网关从AgentCore Identity获取OAuth令牌以进行安全访问。
- 然后，网关与MCP服务器建立安全会话以检索版本能力。
- 最后，网关对MCP服务器的
tools/list端点进行分页调用以检索工具定义，确保网关维护一个当前且准确的工具列表。
MCP目标的ListTools行为
AgentCore Gateway中的ListTools操作提供了一个... [内容被截断]
🚀 想要体验更好更全面的AI调用？
欢迎使用青云聚合API，约为官网价格的十分之一，支持300+全球最新模型，以及全球各种生图生视频模型，无需翻墙高速稳定，文档丰富，小白也可以简单操作。
评论区