全部文档
当前文档
最近更新时间：2026-06-08 16:37:55
MCP是大模型与外部工具和数据源交互的标准协议，随着Agent兴起，管理云产品的方式也发生了变化，金山云支持通过自定义方式将OpenAPI封装为MCP Server，通过自然语言调用和操作金山云产品的OpenAPI。
当前支持自定义方式，按需选择OpenAPI进行封装，在星流平台入口进行部署，可根据业务需要创建多个，满足不同场景
选定的API直接作为tool暴露
可修改API描述和参数
支持跨账号授权
在星流平台-模型应用服务-MCP-MCP管理下，选择“自定义服务”,
点击“新建”，部署方式选择“云产品OpenAPI”；
输入服务名称;
鉴权信息选择API Key或OAuth，默认鉴权方式为OAuth;
鉴权信息 | 入站鉴权 | 支持选择API Key或OAuth，默认为OAuth |
OAuth | 仅选择OAuth鉴权方式时需要选择，支持选择金山云OAuth或自定义OAuth，默认为金山云OAuth | |
MCP账号类型 | 支持选择单账号或多账号，选择多账号时，需配置多账号RAM角色名称 |
添加云产品API，选择需要用到的云产品API版本，添加对应的OpenAPI，为了保证效果，单个MCP Server最多支持30个API；
多账号场景说明：
当客户拥有多个金山云账号，且需要同个一个账号的MCP Server管理多个账号的云资源时，可通过多账号功能实现MCP Server的统一管理。
多账号MCP参数说明
选项 | 说明 |
单账号 | MCP仅能操作当前账号的云端资源 |
多账号 | MCP既能操作当前账号的云端资源，也能够操作其他账号上的云端资源 |
IAM角色管理
为了实现多账号管理，我们通过IAM角色扮演来实现，在被扮演的金山云账号下手动创建IAM角色，并将可信实体设置为用于创建OpenAPI MCP Server的金山云账号。
例如，业务部门为账号A，运维部门为账号B，访问IAM控制台，点击创建角色，在业务部门的账号A下创建一个授权实体类型为云账号，载体信息为其他账号（运维账号B）的IAM角色，并且在运维账号B下创建MCP Server时，指定MCP账号类型为多账号，输入RAM角色名称，即可实现多账号管理
备注：运维账号通过扮演该角色访问业务部门账号的云资源时，需为RAM角色授予相应资源的权限，如何授权，可参考金山云-文档中心-为IAM角色授权
获取MCP Server的服务端点
进入OpenAPI MCP Server，查看基本信息，服务域名即为MCP Server的服务端点
以cherry studio为例，在Cherry Studio客户端添加OAuth方式的应用 不需要手动添加Token请求头，客户端会自动进行认证流程，弹出浏览器认证提交，认证后直接连通
添加后就可以在对话中使用mcp工具调用
此处以Google的ADK框架为例，说明通过OAuth认证实现与OpenAPI MCP Server的连接，并结合大模型与MCP Tools完成对金山云资源的操作。
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools import McpToolset
from google.adk.tools.mcp_tool.mcp_session_manager import create_mcp_http_client
from google.adk.tools.mcp_tool.mcp_toolset import StreamableHTTPConnectionParams
from google.adk.models import LiteLlm
from google.genai import types
import httpx
from mcp.client.auth.oauth2 import OAuthClientProvider, OAuthClientMetadata
from agent.oauth_handler import InMemoryTokenStorage, handle_redirect, handle_callback
from agent.config import config
创建 Runner
self.runner = Runner(
app_name=self.config.agent_name,
agent=self.agent,
session_service=self.session_service
)
logger.info("Agent 设置完成")
async def _create_oauth_provider(self) -> OAuthClientProvider:
"""创建 OAuthClientProvider（使用动态注册或 CIMD），不需显式 client_id/secret"""
client_metadata = OAuthClientMetadata(
client_name=self.config.mcp_oauth_client_name,
redirect_uris=[self.config.mcp_oauth_redirect_uri],
grant_types=["authorization_code", "refresh_token"],
response_types=["code"],
scope=self.config.mcp_oauth_scope,
)
storage = InMemoryTokenStorage()
provider = OAuthClientProvider(
server_url=self.config.mcp_server_url,
client_metadata=client_metadata,
storage=storage,
redirect_handler=handle_redirect,
callback_handler=handle_callback,
)
return provider
async def _get_tools(self) -> list:
"""获取工具列表（使用 OAuthClientProvider 作为 httpx.Auth）"""
tools = []
try:
logger.info(f"正在连接 MCP Server: {self.config.mcp_server_url}")
oauth_provider = await self._create_oauth_provider()
def httpx_client_factory(*args, **kwargs) -> httpx.AsyncClient:
headers = kwargs.pop('headers', None)
timeout = kwargs.pop('timeout', httpx.Timeout(self.config.api_timeout))
return create_mcp_http_client(
headers=headers,
timeout=timeout,
auth=oauth_provider,
)
mcp_connection = StreamableHTTPConnectionParams(
url=self.config.mcp_server_url,
httpx_client_factory=httpx_client_factory,
timeout=self.config.api_timeout,
)
mcp_toolset = McpToolset(connection_params=mcp_connection)
tools.append(mcp_toolset)
logger.info(f"已添加 MCP Toolset: {self.config.mcp_server_url}")
except Exception as e:
logger.error(f"添加 MCP Toolset 失败: {e}", exc_info=True)
return tools
async def run(self, user_input: str, session_id: str = "default"):
"""运行 Agent"""
if not self.runner:
await self.setup()纯净模式