检索网关工具

当您拥有大量工具并且需要为特定用例找到合适工具时，语义搜索尤其重要。开启语义检索后，可以使用tools/call调用网关提供的x_agentarts_gateway_tool_semantic_search工具进行工具检索。

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会使用同一网关资源，调用时请配置此参数，例如：4ada96ff-1e16-4e5b-8fb4-b3f0f207a4f3
        "Mcp-Protocol-Version": "<MCP_PROTOCOL_VERSION>",   # MCP协议版本 
        "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 = "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))