""" IMA 连接器工具 (ima_tool.py) 与腾讯会议不同，IMA 是纯 REST API（HTTP POST + JSON）， 无需外部脚本，直接在进程内调用。 ┌──────────────────────────────────────────────────────┐ │ 绑定流程：侧边栏 inline UI（填写 Client ID + API Key）│ │ 调用流程：SKILL.md + Agent 对话 │ └──────────────────────────────────────────────────────┘ 凭证存储（per-user JSON）： /opt/apps/mindos-next/backend/data/ima_tokens.json 格式：{ "userId": {"client_id": "...", "api_key": "..."} } IMA OpenAPI 文档：https://ima.qq.com/agent-interface """ import json import logging import os import urllib.error import urllib.request from typing import Any, Dict from tools.registry import registry, tool_error, tool_result logger = logging.getLogger(__name__) # ── 常量 ── _BASE_URL = "https://ima.qq.com/" CREDENTIALS_URL = "https://ima.qq.com/agent-interface" _SKILL_VERSION = "1.1.3" # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ # 凭证存取（per-user，wiki/{userId}/.config/ima.json） # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ from tools._user_config import user_config_read, user_config_write, user_config_delete def get_user_credentials(user_id: str) -> Dict[str, str]: """返回 {"client_id": "...", "api_key": "..."} 或空 dict""" creds = user_config_read(user_id, "ima") if creds.get("client_id") and creds.get("api_key"): return creds # 兜底：全局 env cid = os.environ.get("IMA_OPENAPI_CLIENTID", "").strip() key = os.environ.get("IMA_OPENAPI_APIKEY", "").strip() if cid and key: return {"client_id": cid, "api_key": key} return {} def set_user_credentials(user_id: str, client_id: str, api_key: str) -> None: """保存或清除指定用户的 IMA 凭证。两者为空则删除条目。""" client_id = client_id.strip() api_key = api_key.strip() if client_id and api_key: user_config_write(user_id, "ima", {"client_id": client_id, "api_key": api_key}) else: user_config_delete(user_id, "ima") def has_user_credentials(user_id: str) -> bool: return bool(get_user_credentials(user_id)) # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ # HTTP 调用层（进程内，无需子进程） # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ def _ima_call(api_path: str, body: dict, creds: Dict[str, str]) -> Dict[str, Any]: """ 直接发送 HTTP POST 到 IMA OpenAPI。 返回解析后的 JSON dict，出错时返回 {"error": ...}。 """ url = _BASE_URL + api_path.lstrip("/") headers = { "Content-Type": "application/json", "ima-openapi-clientid": creds["client_id"], "ima-openapi-apikey": creds["api_key"], "ima-openapi-ctx": f"skill_version={_SKILL_VERSION}", } data = json.dumps(body).encode("utf-8") req = urllib.request.Request(url, data=data, headers=headers, method="POST") try: with urllib.request.urlopen(req, timeout=30) as r: return json.loads(r.read().decode("utf-8")) except urllib.error.HTTPError as e: body_text = e.read().decode("utf-8", errors="replace")[:500] return {"error": f"HTTP {e.code}", "detail": body_text} except urllib.error.URLError as e: return {"error": f"网络错误: {e}"} except Exception as e: return {"error": str(e)} # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ # ① ima_list_profiles — 连接状态检查 # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ def _ima_list_profiles_handler(args: dict, **kwargs) -> str: user_id = str(args.get("user_id", "")).strip() if not user_id: return tool_error("user_id 参数必填") if not has_user_credentials(user_id): return tool_result( success=True, profiles=[], count=0, credentialsConfigured=False, credentialsUrl=CREDENTIALS_URL, notice="请前往上述链接获取 Client ID 和 API Key，并在侧边栏绑定。", ) creds = get_user_credentials(user_id) # 用 list_note (limit=1) 验证连通性 logger.info("[ImaTool] checking connectivity for user=%s", user_id[:8]) result = _ima_call("openapi/note/v1/list_note", {"cursor": "", "limit": 1}, creds) if "error" in result: status = "error" message = f"连接异常：{result['error']}" else: status = "connected" message = "IMA 已连接。" profiles = [{ "profileName": f"{user_id[:12]}_ima", "label": "IMA 笔记 & 知识库", "status": status, "credentialsConfigured": True, }] return tool_result( success=True, profiles=profiles, count=len(profiles), credentialsConfigured=True, message=message, ) registry.register( name="ima_list_profiles", toolset="connectors", description="检查 IMA 连接状态，返回账号信息。", emoji="📝", handler=_ima_list_profiles_handler, schema={ "name": "ima_list_profiles", "description": "检查 IMA (QQ 笔记/知识库) 是否已连接，返回连接状态。", "parameters": { "type": "object", "properties": { "user_id": {"type": "string", "description": "MINDOS_USER_ID"}, }, "required": ["user_id"], }, }, ) # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ # ② ima_query — 笔记和知识库操作 # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ _ALLOWED_API_PATHS = { # Notes "openapi/note/v1/list_notebook", "openapi/note/v1/list_note", "openapi/note/v1/search_note_book", "openapi/note/v1/get_doc_content", "openapi/note/v1/import_doc", "openapi/note/v1/append_doc", # Knowledge Base "openapi/knowledge/v1/list_knowledge_base", "openapi/knowledge/v1/search_knowledge_base", "openapi/knowledge/v1/create_knowledge_base", "openapi/knowledge/v1/delete_knowledge_base", "openapi/knowledge/v1/add_url", } def _ima_query_handler(args: dict, **kwargs) -> str: user_id = str(args.get("user_id", "")).strip() api_path = str(args.get("api_path", "")).strip() if not user_id: return tool_error("user_id 参数必填") if not api_path: return tool_error("api_path 参数必填") if api_path not in _ALLOWED_API_PATHS: return tool_error( f"不支持的 API 路径：{api_path}。" f"可选：{', '.join(sorted(_ALLOWED_API_PATHS))}" ) if not has_user_credentials(user_id): return tool_error( f"IMA 凭证未配置。请前往 {CREDENTIALS_URL} 获取 Client ID 和 API Key，" "并在侧边栏「IMA」处绑定账号。" ) body = args.get("body", {}) if not isinstance(body, dict): return tool_error("body 必须是 JSON 对象") creds = get_user_credentials(user_id) logger.info("[ImaTool] call %s (user=%s)", api_path, user_id[:8]) result = _ima_call(api_path, body, creds) if "error" in result: return tool_error(f"调用失败：{result['error']}") return tool_result(success=True, data=result) registry.register( name="ima_query", toolset="connectors", description="查询或操作 IMA 笔记和知识库。", emoji="📝", handler=_ima_query_handler, schema={ "name": "ima_query", "description": ( "调用 IMA OpenAPI 操作笔记或知识库。需先在侧边栏完成凭证绑定。\n" "笔记接口示例：\n" " 搜索笔记: api_path=openapi/note/v1/search_note_book, body={\"search_type\":0,\"query_info\":{\"title\":\"xx\"},\"start\":0,\"end\":10}\n" " 读取正文: api_path=openapi/note/v1/get_doc_content, body={\"doc_id\":\"xxx\",\"target_content_format\":0}\n" " 新建笔记: api_path=openapi/note/v1/import_doc, body={\"content\":\"## 标题\\n内容\",\"content_format\":1}\n" " 列出笔记: api_path=openapi/note/v1/list_note, body={\"cursor\":\"\",\"limit\":20}\n" ), "parameters": { "type": "object", "properties": { "user_id": {"type": "string", "description": "MINDOS_USER_ID"}, "api_path": { "type": "string", "description": "IMA API 路径", "enum": sorted(_ALLOWED_API_PATHS), }, "body": { "type": "object", "description": "API 请求体（JSON 对象）", }, }, "required": ["user_id", "api_path"], }, }, )