list_services

Instructions

Implementation Reference

• client-extension/openaaas-mcp-adapter/src/openaaas_mcp_adapter/tools.py:486-537 (handler)

  The MCP tool handler for 'list_services'. Registered via @mcp.tool() decorator inside register_tools(). Calls GET /api/v1/client/services and formats a summary of available agent services.

  @mcp.tool()
  def list_services(server: str = "default") -> str:
      """获取可用的 Agent 服务列表（轻量摘要）"""
      try:
          sc = get_server_config(server)
          api_key = require_api_key(server)
      except RuntimeError as e:
          return f"❌ {e}"
  
      server_url = strip_trailing_slash(sc.get("server_url", ""))
      url = f"{server_url}/api/v1/client/services"
  
      try:
          result = safe_request(
              "GET", url, headers={"Authorization": f"Bearer {api_key}"}
          )
      except OpenAaaSError as e:
          return f"❌ 请求失败: {e}"
  
      services = result if isinstance(result, list) else result.get("services", [])
      if not services:
          return "暂无可用的 Agent 服务"
  
      status_icon = {"online": "🟢", "offline": "🔴", "unknown": "⚪"}
      lines = [f"找到 {len(services)} 个可用服务："]
  
      for i, svc in enumerate(services):
          if not isinstance(svc, dict):
              continue
          svc_id = svc.get("id", "unknown")
          svc_name = svc.get("name", "未命名")
          description = svc.get("description", "无描述")
          agent_status = svc.get("agent_status", "unknown")
          access_type = svc.get("access_type", "unknown")
          has_permission = svc.get("has_permission") is True
          reg_status = svc.get("registration_status")
  
          icon = status_icon.get(agent_status, "⚪")
          perm = "✅ 有权限" if has_permission else "❌ 无权限"
  
          lines.append(f"{i + 1}. {svc_name}")
          lines.append(f"   ID: {svc_id}")
          lines.append(f"   状态: {icon} {agent_status}")
          lines.append(f"   访问类型: {access_type}")
          lines.append(f"   权限: {perm}")
          lines.append(f"   描述: {description}")
          if reg_status:
              lines.append(f"   注册状态: {reg_status}")
          if i < len(services) - 1:
              lines.append("")
  
      return "\n".join(lines)

• client-extension/openaaas-mcp-adapter/src/openaaas_mcp_adapter/tools.py:225-226 (registration)

  The register_tools() function that defines and registers all MCP tools (including list_services) via the @mcp.tool() decorator on the FastMCP instance.

  def register_tools(mcp: FastMCP) -> None:
      """注册所有 OpenAaaS MCP Tools"""

• client-extension/kimi-plugin/list_services.py:16-82 (handler)

  Standalone list_services helper function used by the Kimi plugin, calling GET /api/v1/client/services and returning formatted JSON.

  def list_services(server_url, api_key):
      """
      调用服务列表接口
      
      Args:
          server_url: 服务端基础地址
          api_key: API 密钥
      """
      if not api_key:
          return {"error": "缺少 API Key，请先运行 register 进行注册"}
      
      server_url = server_url.rstrip("/")
      url = f"{server_url}/api/v1/client/services"
      
      try:
          headers = {
              "Authorization": f"Bearer {api_key}"
          }
          req = urllib.request.Request(url, headers=headers, method="GET")
          
          with urllib.request.urlopen(req, timeout=30) as response:
              result = json.loads(response.read().decode("utf-8"))
              
              # 处理返回结果（可能是列表或对象）
              services = result if isinstance(result, list) else result.get("services", [])
              
              # 格式化服务列表
              formatted_services = []
              for svc in services:
                  formatted_services.append({
                      "id": svc.get("id"),
                      "name": svc.get("name"),
                      "description": svc.get("description"),
                      "usage": svc.get("usage", ""),
                      "agent_status": svc.get("agent_status", "unknown"),
                      "access_type": svc.get("access_type", "unknown"),
                      "has_permission": svc.get("has_permission", False),
                      "registration_status": svc.get("registration_status", "unknown")
                  })
              
              return {
                  "content": f"找到 {len(formatted_services)} 个可用服务",
                  "data": {
                      "total": len(formatted_services),
                      "services": formatted_services
                  }
              }
              
      except urllib.error.HTTPError as e:
          error_body = e.read().decode("utf-8")
          try:
              error_data = json.loads(error_body)
              error_msg = error_data.get("error") or error_data.get("message") or error_body
          except:
              error_msg = error_body or e.reason
          
          if e.code == 401:
              return {"error": f"认证失败 (401): API Key 无效，请检查 config.json 中的 api_key 是否正确"}
          elif e.code == 403:
              return {"error": f"权限不足 (403): 无权访问服务列表"}
          return {"error": f"请求失败 (HTTP {e.code}): {error_msg}"}
      except urllib.error.URLError as e:
          return {"error": f"连接失败: {e.reason}，请检查服务端地址是否正确"}
      except json.JSONDecodeError as e:
          return {"error": f"JSON 解析错误: {str(e)}"}
      except Exception as e:
          return {"error": f"请求失败: {str(e)}"}

• dash/src/aaas_dashboard/client.py:653-662 (helper)

  Admin dashboard client method list_services() that calls GET /api/v1/services (admin endpoint) to list all services.

  def list_services(self) -> list[dict]:
      """List all services (admin perspective)."""
      try:
          data = self._get("/api/v1/services")
          if isinstance(data, list):
              return data
          return data.get("services", [])
      except requests.RequestException as e:
          print(f"[ERROR] Network error in list_services: {e}")
          return []

• dash/src/aaas_dashboard/client.py:1027-1029 (helper)

  Mock client method list_services() returning mock services data for dashboard testing.

  def list_services(self) -> list[dict]:
      """List mock services."""
      return [dict(s) for s in self._services]