MCP Gateway SDK
场景介绍
MCP Gateway SDK提供管理MCP网关及相关操作的命令行工具,支持网关和网关目标的创建、查询、更新、删除等功能。
前提条件
- 已开通AgentArts。
- 登录控制台获取AK/SK,详情请参考认证鉴权。
export HUAWEICLOUD_SDK_AK="your-access-key" export HUAWEICLOUD_SDK_SK="your-secret-key"
- 已安装Python,且版本不低于3.10。查看Python版本的命令示例:
python --version
网关管理
- 创建MCP Gateway
def create_mcp_gateway( self, name: Optional[str] = None, description: Optional[str] = None, protocol_type: Optional[str] = "mcp", authorizer_type: Optional[str] = "iam", agency_name: Optional[str] = None, authorizer_configuration: Optional[CoreGatewayAuthorizerConfiguration] = None, log_delivery_configuration: Optional[CoreGatewayLogDeliveryConfigurationRequestBody] = None, outbound_network_configuration: Optional[CoreGatewayOutboundNetworkConfiguration] = None, skip_ssl_verification: Optional[bool] = False ) -> RequestResult表1 创建MCP Gateway参数说明 参数
类型
是否必选
默认值
说明
name
Optional[str]
否
gateway-{8个随机字符后缀}
网关名称
description
Optional[str]
否
-
网关的详细描述
protocol_type
Optional[str]
否
mcp
网关协议类型
authorizer_type
Optional[str]
否
iam
授权器类型,可选值:custom_jwt、iam、api_key
agency_name
Optional[str]
否
AgentArksCoreGateway
网关委托身份的委托名称
authorizer_configuration
Optional[CoreGatewayAuthorizerConfiguration]
否
-
授权器配置
log_delivery_configuration
Optional[CoreGatewayLogDeliveryConfigurationRequestBody]
否
-
日志投递配置
outbound_network_configuration
Optional[CoreGatewayOutboundNetworkConfiguration]
否
-
出站网络配置
授权器配置要求:
- 当authorizer_type = 'custom_jwt' 时,必须提供custom_jwt_authorizer配置。
- 当authorizer_type = 'api_key' 时,必须提供key_auth配置。
- 查询MCP Gateway信息
def get_mcp_gateway(self, gateway_id: str) -> ShowCoreGatewayResponse
表2 查询MCP Gateway参数说明 参数
类型
是否必选
说明
gateway_id
str
是
要查询的Gateway的唯一标识符
- 更新指定的MCP Gateway信息
def update_mcp_gateway( self, gateway_id: str, description: Optional[str] = None, log_delivery_configuration: Optional[CoreGatewayLogDeliveryConfigurationRequestBody] = None, ) -> RequestResult表3 更新MCP Gateway参数说明 参数
类型
是否必选
说明
gateway_id
str
是
要更新的Gateway的唯一标识符
description
Optional[str]
否
网关的详细描述
log_delivery_configuration
Optional[CoreGatewayLogDeliveryConfigurationRequestBody]
否
日志投递配置
更新操作为增量更新,只传入需要更新的字段,未传入的字段将保持原值不变。
- 删除指定的MCP Gateway信息
def delete_mcp_gateway(self, gateway_id: str) -> DeleteCoreGatewayResponse
表4 更新MCP Gateway参数说明 参数
类型
是否必选
说明
gateway_id
str
是
要删除的Gateway的唯一标识符
- 获取MCP Gateway列表
def list_mcp_gateways( self, name: Optional[str] = None, status: Optional[str] = None, gateway_id: Optional[str] = None, limit: Optional[int] = None, offset: Optional[int] = None) -> ListCoreGatewaysResponse表5 获取MCP Gateway实例列表参数说明 参数
类型
是否必选
默认值
说明
name
Optional[str]
否
-
按名称过滤网关,支持模糊匹配
status
Optional[str]
否
-
按状态过滤网关
gateway_id
Optional[str]
否
-
按网关ID过滤,支持多个ID(最多20个),用逗号分隔
limit
Optional[int]
否
50
返回结果的最大数量
offset
Optional[int]
否
0
返回结果的偏移量
- 在指定的MCP Gateway下创建新的Target实例
def create_mcp_gateway_target( self, gateway_id: str, name: Optional[str] = None, description: Optional[str] = None, target_configuration: Optional[CoreGatewayTargetConfiguration] = None, credential_provider_configuration: Optional[CoreGatewayCredentialProviderConfiguration] = None) -> CreateCoreGatewayTargetResponse表6 创建MCP Target参数说明 参数
类型
是否必选
默认值
说明
gateway_id
str
是
-
目标Gateway的唯一标识符
name
Optional[str]
否
target-{8个随机字符后缀}
Target服务名称
description
Optional[str]
否
-
Target服务描述
target_configuration
Optional[CoreGatewayTargetConfiguration]
是
-
Target服务配置,mcp_server内容必须提供
credential_provider_configuration
Optional[CoreGatewayCredentialProviderConfiguration]
否
-
凭证提供者配置
凭证提供者配置要求:
- 如果不提供,credential_provider_type默认为none,表示无认证。
- 当credential_provider_type='oauth' 时,必须提供oauth_credential_provider。
- 当credential_provider_type='api_key' 时,必须提供api_key_credential_provider。
- 查询指定Gateway下的特定Target实例的详细信息
def get_mcp_gateway_target(self, gateway_id: str, target_id: str) -> ShowCoreGatewayTargetResponse
表7 查询MCP Target信息参数说明 参数
类型
是否必选
说明
gateway_id
str
是
Target所属Gateway的唯一标识符
target_id
str
是
要查询的Target的唯一标识符
- 更新指定Gateway下的特定Target实例
def update_mcp_gateway_target( self, gateway_id: str, target_id: str, name: Optional[str] = None, description: Optional[str] = None, target_configuration: Optional[CoreGatewayTargetConfiguration] = None, credential_provider_configuration: Optional[CoreGatewayCredentialProviderConfiguration] = None) -> UpdateCoreGatewayTargetResponse表8 更新MCP Target参数信息 参数
类型
是否必选
说明
gateway_id
str
是
Target所属Gateway的唯一标识符
target_id
str
是
要更新的Target的唯一标识符
name
Optional[str]
否
Target服务名称
description
Optional[str]
否
Target服务描述
target_configuration
Optional[CoreGatewayTargetConfiguration]
否
Target服务配置
credential_provider_configuration
Optional[CoreGatewayCredentialProviderConfiguration]
否
凭证提供者配置
- 删除指定Gateway下的特定Target实例
def delete_mcp_gateway_target(self, gateway_id: str, target_id: str) -> DeleteCoreGatewayTargetResponse
表9 删除MCP Target参数说明 参数
类型
是否必选
说明
gateway_id
str
是
Target所属Gateway的唯一标识符
target_id
str
是
要删除的Target的唯一标识符
- 查询指定Gateway下的所有Target列表
def list_mcp_gateway_targets( self, gateway_id: str, limit: Optional[int] = None, offset: Optional[int] = None) -> ListCoreGatewayTargetsResponse表10 查询指定Gateway下的Target列表参数说明 参数
类型
是否必选
默认值
说明
gateway_id
str
是
-
要查询Targets的Gateway的唯一标识符
limit
Optional[int]
否
50
每页返回的最大结果数量
offset
Optional[int]
否
0
返回结果的偏移量
参数说明
- 请求参数
表11 创建MCP Gateway参数 参数
类型
说明
name
String
网关名称,如果不提供则默认为gateway-{8个随机字符后缀}
description
String
网关的详细描述
protocol_type
String
协议类型,默认为 "MCP"
authorizer_type
String
授权器类型,可选值:custom_jwt、iam、api_key
agency_name
String
委托名称,默认为 "AgentArksCoreGateway"
authorizer_configuration
Object
授权器配置对象
log_delivery_configuration
Object
日志投递配置对象
outbound_network_configuration
Object
出站网络配置对象
表12 Target配置对象 参数
类型
说明
mcp_server
Object
MCP服务器配置
表13 凭证提供者配置对象 参数
类型
说明
credential_provider_type
String
凭证提供者类型
credential_provider
Object
凭证提供者配置
- 响应参数
表14 CreateCoreGatewayResponse 字段
类型
说明
gateway_id
String
网关的唯一标识符
name
String
网关名称
description
String
网关的详细描述
status
String
网关当前状态
protocol_type
String
网关协议类型
authorizer_type
String
授权器类型
authorizer_configuration
Object
授权器配置
endpoint_url
String
网关访问的URL端点
log_delivery_configuration
Object
日志投递配置
workload_identity
Object
工作负载身份标识符
outbound_network_configuration
Object
网络配置
agency_name
String
网关委托名称
created_at
String
网关创建时间戳
updated_at
String
网关最后更新时间戳
表15 ListCoreGatewaysResponse 字段
类型
说明
gateways
List[ShowCoreGatewayResponse]
网关详情列表
size
Integer
当前页返回的网关数量
total
Integer
网关总数量
表16 ShowCoreGatewayTargetResponse 字段
类型
说明
target_id
String
目标唯一标识符
name
String
目标名称
description
String
目标描述
status
String
目标状态
connection_status
String
连接状态
target_configuration
Object
目标配置
credential_provider_configuration
Object
凭证提供者配置
created_at
String
创建时间戳
updated_at
String
更新时间戳
快速入门
- 创建网关。
# Client模式 result = client.create_mcp_gateway( name="my-gateway", description="我的第一个MCP网关" ) gateway_id = result.data.get("gateway_id") # CLI模式 agentarts mcp-gateway create-mcp-gateway --name "my-gateway" --description "我的第一个MCP网关" - 创建目标。
# Client模式 target_result = client.create_mcp_gateway_target( gateway_id=gateway_id, target_configuration={ "mcp_server": { "endpoint": "https://echo.example.com", "server_type": "sse" } } ) target = target_result.data.get("target") target_id = target.get("target_id") # CLI模式 agentarts mcp-gateway create-mcp-gateway-target <gateway-id> \ --target-configuration '{\"mcp_server\":{\"endpoint\":\"https://example.com/mcp\",\"server_type\":\"sse\"}}' - 查询信息。
# Client模式 gateway_info = client.get_mcp_gateway(gateway_id) target_info = client.get_mcp_gateway_target(gateway_id, target_id) # CLI模式 agentarts mcp-gateway get-mcp-gateway <gateway-id> agentarts mcp-gateway get-mcp-gateway-target <gateway-id> <target-id>
示例代码
from typing import Optional, List
from agentarts.sdk.mcpgateway import MCPGatewayClient
# 初始化客户端
client = MCPGatewayClient()
# 创建MCP Gateway
result = client.create_mcp_gateway(
name="production-echo-gateway",
description="生产环境 Echo 服务网关",
authorizer_type="iam"
)
gateway = result.data
print(f"创建的网关ID: {gateway.get('gateway_id')}")
# 更新网关描述
result = client.update_mcp_gateway(
gateway_id=gateway.get('gateway_id'),
description="更新后的网关描述"
)
updated_gateway = result.data
print(f"更新后的描述: {updated_gateway.get('description')}")
# 在网关下创建Target(MCP服务器类型)
result = client.create_mcp_gateway_target(
gateway_id=gateway.get('gateway_id'),
name="echo-service-target",
description="Echo服务目标",
target_configuration={
"mcp_server": {
"endpoint": "https://echo.example.com",
"server_type": "sse"
}
}
)
target = result.data.get("target")
target_id = target.get("target_id")
print(f"目标创建成功,ID: {target_id}")
# 在网关下创建Target(OpenAPI内联文档类型)
result = client.create_mcp_gateway_target(
gateway_id=gateway.get('gateway_id'),
name="petstore-api",
description="PetStore API目标",
target_configuration={
"openapi": {
"payload": "{\"info\":{\"description\":\"Search & Content Management API (OpenAPI 3.0.0 compliant)\",\"title\":\"Search & Content API - APIs Scene\",\"version\":\"1.0.0\"},\"openapi\":\"3.0.0\",\"servers\":[{\"url\":\"https://api.example.com/v1\"}],\"paths\":{\"/api/resource/1\":{\"post\":{\"operationId\":\"createResource1\",\"summary\":\"Create resource 1\",\"requestBody\":{\"required\":true,\"content\":{\"application/json\":{\"schema\":{\"$ref\":\"#/components/schemas/ContentBase\"}}}},\"responses\":{\"201\":{\"description\":\"Created success\",\"content\":{\"application/json\":{\"schema\":{\"type\":\"object\",\"properties\":{\"id\":{\"type\":\"integer\"},\"title\":{\"type\":\"string\"}}}}}}}}}},\"components\":{\"schemas\":{\"ContentBase\":{\"type\":\"object\",\"required\":[\"title\"],\"properties\":{\"title\":{\"type\":\"string\"},\"isPublic\":{\"type\":\"boolean\"},\"tags\":{\"type\":\"array\",\"items\":{\"type\":\"string\"}},\"metadata\":{\"type\":\"object\",\"properties\":{\"createTime\":{\"type\":\"string\"},\"operator\":{\"type\":\"string\"}}}}}}}}"
}
}
)
target = result.data.get("target")
target_id = target.get("target_id")
print(f"OpenAPI目标创建成功,ID: {target_id}")
# 在网关下创建Target(OpenAPI OBS文档类型)
result = client.create_mcp_gateway_target(
gateway_id=gateway.get('gateway_id'),
name="petstore-api-obs",
description="PetStore API目标(OBS)",
target_configuration={
"openapi": {
"obs": {
"bucket_name": "api-specs-bucket",
"object_key": "specs/petstore.yaml"
}
}
}
)
target = result.data.get("target")
target_id = target.get("target_id")
print(f"OpenAPI OBS 目标创建成功,ID: {target_id}")
# 查询网关信息
result = client.get_mcp_gateway(gateway.get('gateway_id'))
gateway_info = result.data
print(f"网关状态: {gateway_info.get('status')}")
# 查询目标信息
result = client.get_mcp_gateway_target(gateway.get('gateway_id'), target.get('target_id'))
target_info = result.data.get("target")
print(f"目标状态: {target_info.get('status')}")
# 列出网关下的所有目标
result = client.list_mcp_gateway_targets(gateway.get('gateway_id'))
targets = result.data
print(f"网关下有 {targets.get('total')} 个目标")
# 更新Target描述
result = client.update_mcp_gateway_target(
gateway_id=gateway.get('gateway_id'),
target_id=target.get('target_id'),
description="更新后的Echo服务目标描述"
)
updated_target = result.data.get("target")
print(f"更新后的描述: {updated_target.get('description')}")
# 删除Target
client.delete_mcp_gateway_target(gateway.get('gateway_id'), target.get('target_id'))
print("目标已删除")
# 删除Gateway
client.delete_mcp_gateway(gateway.get('gateway_id'))
print("网关已删除")