在智能体中使用网关

创建网关后，您需要将已经创建的网关集成到智能体的代码中或添加到已有的MCP智能体客户端，以便后续调用网关能力，可以参考以下示例代码，并根据需求自定义修改其中的信息，生成实际使用的代码。

前提条件

已完成创建网关。

在智能体中使用网关

登录AgentArts智能体开发平台。

在左侧导航栏选择“开发中心 > 组件库 ”，单击“网关”页签，进入网关界面。

在网关列表单击操作列的“调用代码”，查看并复制代码示例，并根据需求自定义修改其中的信息，生成实际使用的代码。

您可以在请求体中指定tools/list作为请求方法获取网关提供的所有可用工具。

您可以在请求体中指定tools/call作为请求方法用于调用特定的工具。

API Key认证的tools/list调用示例如下：

import requests
import json
import re


def list_tools(gateway_url, access_token):
    headers = {
        "Content-Type": "application/json",
        "Accept": "application/json, text/event-stream",
        "mcp-session-id": "<MCP_SESSION_ID>",  # MCP会话ID，相同的sessionid会使用同一网关资源
        "Authorization": f"Bearer {access_token}"      
    }
    payload = {"jsonrpc": "2.0","id": "list-tools-request","method": "tools/list"}
    response = requests.post(gateway_url, headers=headers, json=payload) # 若需要忽略ssl校验，可补充参数verify=False
    raw = response.content.decode('utf-8', errors='replace')
    ct = response.headers.get('content-type', '')
    if 'text/event-stream' in ct:
        for line in raw.strip().split('\n'):
            if line.startswith('data:'):
                try:
                    return json.loads(line[5:].strip())
                except json.JSONDecodeError:
                    continue
        m = re.search(r'\{.*\}', raw, re.DOTALL)
        if m:
            return json.loads(m.group())
    try:
        return json.loads(raw)
    except json.JSONDecodeError:
        print(f"[Fail]: {raw[:200]}")
        return None

gateway_url = "<GATEWAY_URL>"   # 网关接口地址（业务API调用入口）
access_token = "<API_KEY>"      # 认证凭证：APIKey认证直接填写密钥；无认证则留空/删除
result = list_tools(gateway_url, access_token)
if result:
    print(json.dumps(result, indent=2, ensure_ascii=False))

API Key认证的tools/call调用示例如下：

import requests
import json
import re


def call_tool(gateway_url, access_token, tool_name, arguments):
    headers = {
        "Content-Type": "application/json",
        "Accept": "application/json, text/event-stream",
        "mcp-session-id": "<MCP_SESSION_ID>",  # MCP会话ID，相同的sessionid会使用同一网关资源
        "Authorization": f"Bearer {access_token}"
    }
    payload = {"jsonrpc": "2.0","id": "call-tool-request","method": "tools/call","params": {"name": tool_name,"arguments": arguments}}
    response = requests.post(gateway_url, headers=headers, json=payload) # 若需要忽略ssl校验，可补充参数verify=False
    raw = response.content.decode('utf-8', errors='replace')
    ct = response.headers.get('content-type', '')
    if 'text/event-stream' in ct:
        for line in raw.strip().split('\n'):
            if line.startswith('data:'):
                try:
                    return json.loads(line[5:].strip())
                except json.JSONDecodeError:
                    continue
        m = re.search(r'\{.*\}', raw, re.DOTALL)
        if m:
            return json.loads(m.group())
    try:
        return json.loads(raw)
    except json.JSONDecodeError:
        print(f"[Fail]: {raw[:200]}")
        return None

gateway_url = "<GATEWAY_URL>"   # 网关接口地址（业务API调用入口）
access_token = "<API_KEY>"      # 认证凭证：APIKey认证直接填写密钥；无认证则留空/删除
tool_name = "<TOOL_NAME>"        # 待调用的工具名称，通过 tools/list 接口查询获取
arguments = {"param1": "value1", "param2": "value2"}                 # 工具调用入参，严格按照工具定义的参数结构填写
result = call_tool(gateway_url, access_token, tool_name, arguments)
if result:
    print(json.dumps(result, indent=2, ensure_ascii=False))

OAuth认证的tools/list调用示例如下：

import requests
import json
import re


def list_tools(gateway_url, access_token):
    headers = {
        "Content-Type": "application/json",
        "Accept": "application/json, text/event-stream",
        "mcp-session-id": "<MCP_SESSION_ID>",     # MCP会话ID，相同的sessionid会使用同一网关资源
        "Authorization": f"Bearer {access_token}"
    }
    payload = {"jsonrpc": "2.0", "id": "list-tools-request", "method": "tools/list"}
    response = requests.post(gateway_url, headers=headers, json=payload) # 若需要忽略ssl校验，可补充参数verify=False
    raw = response.content.decode('utf-8', errors='replace')
    ct = response.headers.get('content-type', '')
    if 'text/event-stream' in ct:
        for line in raw.strip().split('\n'):
            if line.startswith('data:'):
                try:
                    return json.loads(line[5:].strip())
                except json.JSONDecodeError:
                    continue
        m = re.search(r'\{.*\}', raw, re.DOTALL)
        if m:
            return json.loads(m.group())
    try:
        return json.loads(raw)
    except json.JSONDecodeError:
        print(f"[Fail]: {raw[:200]}")
        return None

def fetch_access_token(token_url, client_id, client_secret):
    response = requests.post(
        token_url,
        data=f"grant_type=client_credentials&client_id={client_id}&client_secret={client_secret}",  # scope为非必填参数，scope表示你要申请的权限范围
        headers={'Content-Type': 'application/x-www-form-urlencoded'}
    )
    return response.json()['access_token']

gateway_url = "<GATEWAY_URL>"               # 网关接口地址（业务API调用入口）
token_url = "<TOKEN_URL>"                   # Oauth2.0 令牌获取地址
client_id = "<CLIENT_ID>"                   # 客户端ID
client_secret = "<CLIENT_SECRET>"           # 客户端密钥（认证凭证）
access_token = fetch_access_token(token_url, client_id, client_secret)      # 接口调用鉴权令牌（有效期内通用）
result = list_tools(gateway_url, access_token)
if result:
    print(json.dumps(result, indent=2, ensure_ascii=False))

IAM认证的tools/list调用示例如下，API签名详情请参见API签名指南：

import requests
import json
import re


def list_tools(gateway_url, access_token):
    headers = {
        "Content-Type": "application/json",
        "Accept": "application/json, text/event-stream",
        "mcp-session-id": "<MCP_SESSION_ID>",     # MCP会话ID，相同的sessionid会使用同一网关资源
        "X-Sdk-Date": "<X_SDK_DATE>",             # 网关鉴权时间戳
        "X-Sdk-Content-Sha256": "UNSIGNED-PAYLOAD",
        "Authorization": access_token                       
    }
    payload = {"jsonrpc": "2.0", "id": "list-tools-request", "method": "tools/list"}
    response = requests.post(gateway_url, headers=headers, json=payload) # 若需要忽略ssl校验，可补充参数verify=False
    raw = response.content.decode('utf-8', errors='replace')
    ct = response.headers.get('content-type', '')
    if 'text/event-stream' in ct:
        for line in raw.strip().split('\n'):
            if line.startswith('data:'):
                try:
                    return json.loads(line[5:].strip())
                except json.JSONDecodeError:
                    continue
        m = re.search(r'\{.*\}', raw, re.DOTALL)
        if m:
            return json.loads(m.group())
    try:
        return json.loads(raw)
    except json.JSONDecodeError:
        print(f"[Fail]: {raw[:200]}")
        return None

gateway_url = "<GATEWAY_URL>"       # 网关接口地址（业务API调用入口）
access_token = "<AUTHORIZATION>"    # 鉴权Authorization
result = list_tools(gateway_url, access_token)
if result:
    print(json.dumps(result, indent=2, ensure_ascii=False))

父主题： 网关

上一篇：在网关中创建Target

下一篇：查看网关日志

意见反馈

文档内容是否对您有帮助？

有帮助没帮助

提供反馈

提交成功！非常感谢您的反馈，我们会继续努力做到更好！您可在我的云声建议查看反馈及问题处理状态。

系统繁忙，请稍后重试

如您有其它疑问，您也可以通过华为云社区问答频道来与我们联系探讨

云宝助手提问云社区提问

在智能体中使用网关

前提条件

在智能体中使用网关

相关文档

意见反馈

文档内容是否对您有帮助？