lidf
69dd868e2f
- 独立 pyproject.toml（pip install -e .） - vendor_hermes.sh 已改为显式路径模式（不再依赖相对目录） - 包含 hermes vendor 快照
255 lines
9.3 KiB
Python
255 lines
9.3 KiB
Python
"""
|
||
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"],
|
||
},
|
||
},
|
||
)
|