更新时间:2026-07-14 gmt 08:00

在智能体中使用网关-j9九游会登录

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

前提条件

完成

在智能体中使用网关

  1. 登录。
  1. 在左侧导航栏选择“开发中心 > 组件库 ”,单击“网关”页签,进入网关界面。
  1. 在网关列表单击操作列的“调用代码”,查看并复制代码示例,并根据需求自定义修改其中的信息,生成实际使用的代码。

    您可以在请求体中指定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会话id,相同的sessionid会使用同一网关资源调用时请配置此参数,例如:4ada96ff-1e16-4e5b-8fb4-b3f0f207a4f3
        "mcp-protocol-version": "",   # mcp协议版本,2025-06-18版本新增请求头          
        "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 = ""   # 网关接口地址(业务api调用入口)
access_token = ""      # 认证凭证: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会话id,相同的sessionid会使用同一网关资源调用时请配置此参数,例如:4ada96ff-1e16-4e5b-8fb4-b3f0f207a4f3
        "mcp-protocol-version": "",   # mcp协议版本,2025-06-18版本新增请求头 
        "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 = ""   # 网关接口地址(业务api调用入口)
access_token = ""      # 认证凭证:apikey认证直接填写密钥;无认证则留空/删除
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))

iam认证的tools/list调用示例

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会话id,相同的sessionid会使用同一网关资源调用时请配置此参数,例如:4ada96ff-1e16-4e5b-8fb4-b3f0f207a4f3
        "mcp-protocol-version": "",   # mcp协议版本,2025-06-18版本新增请求头 
        "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 = ""       # 网关接口地址(业务api调用入口)
access_token = ""    # 鉴权authorization
result = list_tools(gateway_url, access_token)
if result:
    print(json.dumps(result, indent=2, ensure_ascii=false))

api key认证的ping的调用示例

import requests
import json
import re
def ping(gateway_url, access_token):
    headers = {
        "content-type": "application/json",
        "accept": "application/json, text/event-stream",
        "mcp-session-id": "",  # mcp会话id,相同的sessionid会使用同一网关资源调用时请配置此参数,例如:4ada96ff-1e16-4e5b-8fb4-b3f0f207a4f3
        "mcp-protocol-version": "",   # mcp协议版本,2025-06-18版本新增请求头 
        "authorization": f"bearer {access_token}"
    }
    payload = {"jsonrpc": "2.0", "id": "ping-request", "method": "ping"}
    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 = ""   # 网关接口地址(业务api调用入口)
access_token = ""      # 认证凭证:apikey认证直接填写密钥;无认证则留空/删除
result = ping(gateway_url, access_token)
if result:
    print(json.dumps(result, indent=2, ensure_ascii=false))

语义检索的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会话id,相同的sessionid会使用同一网关资源调用时请配置此参数,例如:4ada96ff-1e16-4e5b-8fb4-b3f0f207a4f3
        "mcp-protocol-version": "",   # mcp协议版本,2025-06-18版本新增请求头 
        "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 = ""   # 网关接口地址(业务api调用入口)
access_token = ""      # 认证凭证:apikey认证直接填写密钥;无认证则留空/删除
tool_name = "x_agentarts_gateway_tool_semantic_search"    # 语义检索调用的工具名称,固定为 "x_agentarts_gateway_tool_semantic_search"
arguments = {"query": "value"}        # 语义检索调用入参,value表示检索提示词
result = call_tool(gateway_url, access_token, tool_name, arguments)
if result:
    print(json.dumps(result, indent=2, ensure_ascii=false))

相关文档

网站地图