# 使用网关
创建网关后，您需要将已经创建的网关集成到智能体的代码中或添加到已有的智能体，以便后续调用网关能力。
#### 前提条件
- 已完成[创建网关](https://support.huaweicloud.com/highcode-agentarts/agentarts_10_073.html)。
- 网关提供入站认证鉴权和出站身份认证两层安全机制，保障调用方与后端工具之间的安全通信。调用网关或调用网关Target需要获取对应凭证：
  - 入站身份认证：调用方访问网关时的认证和鉴权，用于验证调用方身份并控制访问权限。详情请参见[认证鉴权](https://support.huaweicloud.com/api-agentarts/agentarts_07_0005.html)。
  
  - 出站身份认证：网关访问后端工具服务时的身份认证，用于配置网关以特定身份访问后端资源。详情请参见[出站身份认证](https://support.huaweicloud.com/highcode-agentarts/agentarts_10_025.html#section2)。
   
 
#### 使用网关
1. 登录[AgentArts智能体平台](https://console.huaweicloud.com/agentarts/#/home/overview)。

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

3. 在网关列表单击操作列的"调用代码"，查看并复制代码示例，并根据需求自定义修改其中的信息。 
   您可以在请求体中指定tools/list作为请求方法获取网关提供的所有可用工具。
   您可以在请求体中指定tools/call作为请求方法用于调用特定的工具。
   
   
 
#### 列出网关工具
在请求体中指定tools/list作为请求方法，获取网关提供的所有可用工具列表。返回结果包含工具名称、描述和参数定义等信息，可用于后续工具调用。
请求体示例：
```
{
  "jsonrpc": "2.0",
  "id": "list-tools-request",
  "method": "tools/list",
  "params": {
    "cursor": "<CURSOR>"
  }
}
```
其中cursor为可选参数，用于分页请求，首次请求无需携带。
响应体示例：
```
{
  "jsonrpc": "2.0",
  "id": "list-tools-request",
  "result": {
    "tools": [
      {
        "name": "tool_name",
        "description": "工具描述",
        "inputSchema": {}
      }
    ],
    "nextCursor": "<NEXT_CURSOR>"
  }
}
```
网关支持分页返回工具列表。当Target响应tools/list请求返回nextCursor字段且非空时，表示该Target有更多工具未返回，需要将nextCursor的值作为下一次请求的cursor参数继续请求，重复此过程直至nextCursor为空，即可获取完整的工具列表，详情请参见[MCP协议分页介绍](https://modelcontextprotocol.io/specification/draft/server/utilities/pagination)。
网关会根据不同的Target进行分页，每个Target的工具列表独立分页返回。即每次返回的工具属于同一Target，获取下一个Target的工具需要继续分页请求。每个Target最多返回1000个工具。
- 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会使用同一网关资源，调用时请配置此参数，例如：4ada96ff-1e16-4e5b-8fb4-b3f0f207a4f3
          "Mcp-Protocol-Version": "<MCP_PROTOCOL_VERSION>",   # MCP协议版本          
          "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))
  ```
  
- IAM认证的tools/list调用示例
  API签名详情请参见[API签名指南](https://support.huaweicloud.com/devg-apisign/api-sign-provide01.html)：
  ```
  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会使用同一网关资源，调用时请配置此参数，例如：4ada96ff-1e16-4e5b-8fb4-b3f0f207a4f3
          "Mcp-Protocol-Version": "<MCP_PROTOCOL_VERSION>",   # MCP协议版本 
          "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))
  ```
  
#### 调用网关工具
在请求体中指定tools/call作为请求方法，调用网关提供的特定工具。调用时需指定工具名称和入参，工具名称可通过tools/list接口查询获取，入参需严格按照工具定义的参数结构填写。
请求体示例：
```
{
  "jsonrpc": "2.0",
  "id": "call-tool-request",
  "method": "tools/call",
  "params": {
    "name": "<TOOL_NAME>",
    "arguments": {
      "param1": "value1",
      "param2": "value2"
    }
  }
}
```
其中name为待调用的工具名称，通过tools/list接口查询获取；arguments为工具调用入参，需严格按照工具定义的参数结构填写。
响应体示例：
```
{
  "jsonrpc": "2.0",
  "id": "call-tool-request",
  "result": {
    "content": [
      {
        "type": "text",
        "text": "工具调用结果"
      }
    ]
  }
}
```
- 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会使用同一网关资源，调用时请配置此参数，例如：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 = "<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/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会使用同一网关资源，调用时请配置此参数，例如：4ada96ff-1e16-4e5b-8fb4-b3f0f207a4f3
          "Mcp-Protocol-Version": "<MCP_PROTOCOL_VERSION>",   # MCP协议版本 
          "X-Sdk-Date": "<X_SDK_DATE>",             # 网关鉴权时间戳
          "X-Sdk-Content-Sha256": "UNSIGNED-PAYLOAD",
          "Authorization": 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 = "<AUTHORIZATION>"    # 鉴权Authorization
  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))
  ```
  
 #### 检索网关工具
当您拥有大量工具并且需要为特定用例找到合适工具时，语义搜索尤其重要。开启语义检索后，可以使用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))
```
#### 其他示例
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_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": "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 = "<GATEWAY_URL>"   # 网关接口地址（业务API调用入口）
access_token = "<API_KEY>"      # 认证凭证：APIKey认证直接填写密钥；无认证则留空/删除
result = ping(gateway_url, access_token)
if result:
    print(json.dumps(result, indent=2, ensure_ascii=False))
```