from mcp.server.fastmcp import FastMCP import os import json import uuid import asyncio import logging import websockets import urllib.request import urllib.error import urllib.parse import datetime from typing import Any # Compatibility patch for newer mcp/pydantic combinations that may fail to # build output schemas for plain return annotations such as `-> str`. try: import mcp.server.fastmcp.utilities.func_metadata as _fm from pydantic import create_model as _create_model def _patched_create_wrapped_model(func_name: str, annotation: Any) -> Any: return _create_model(f"{func_name}Output", **{"result": (annotation, ...)}) _fm._create_wrapped_model = _patched_create_wrapped_model except Exception: pass logging.basicConfig(level=logging.INFO) logger = logging.getLogger("openclaw_mcp") BRIDGE_HOST = os.getenv("BRIDGE_HOST", "0.0.0.0") BRIDGE_PORT = int(os.getenv("BRIDGE_PORT", "8765")) mcp = FastMCP("OpenClaw Assistant", host=BRIDGE_HOST, port=BRIDGE_PORT) OPENCLAW_HTTP = os.getenv("OPENCLAW_URL", "http://192.168.50.139:18789").rstrip("/") OPENCLAW_TOKEN = os.getenv("OPENCLAW_TOKEN", "").strip() OPENCLAW_WS_URL = OPENCLAW_HTTP.replace("http://", "ws://").replace("https://", "wss://") + "/ws" HA_URL = os.getenv("HA_URL", "http://192.168.50.139:8123").rstrip("/") HA_TOKEN = os.getenv("HA_TOKEN", "").strip() YOUTUBE_API_KEY = os.getenv("YOUTUBE_API_KEY", "").strip() MINIMAX_API_KEY = os.getenv("MINIMAX_API_KEY", "").strip() LOG_FILE = "/tmp/ha_mcp_debug.log" def _log(msg): with open(LOG_FILE, "a") as f: f.write(f"{datetime.datetime.now()} {msg}\n") # ==================== 长期记忆 ==================== MEMORY_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "memories.json") def _load_memories(): if not os.path.exists(MEMORY_FILE): return [] try: with open(MEMORY_FILE, "r", encoding="utf-8") as f: return json.load(f) except Exception: return [] def _save_memories(memories): with open(MEMORY_FILE, "w", encoding="utf-8") as f: json.dump(memories, f, ensure_ascii=False, indent=2) @mcp.tool() def save_memory(content: str, category: str = "general") -> str: """ 保存一条长期记忆，永久保存。以后的对话都能记住。 什么时候应该调用: - 用户说"记住"、"帮我记一下"、"别忘了"、"下次记得"等 - 用户告诉你购物清单、待办事项、日程安排等需要记住的事 - 用户纠正了你的错误（记住正确做法） - 你学到了新的设备使用经验 - 用户表达了偏好（喜欢什么、不喜欢什么） 应该保存的内容举例: - 待办/购物："要买牛奶和鸡蛋"、"周末要去修车" - 设备经验："电视播放YouTube要用 media_player.dian_shi" - 用户偏好："用户喜欢在客厅display上看YouTube" - 纠错："客厅灯是switch类型不是light类型" - 任何用户希望你记住的信息 参数: - content: 记忆内容，写清楚具体信息 - category: 分类，可选 device（设备知识）、preference（用户偏好）、experience（使用经验）、todo（待办/购物）、general（其他） """ _log(f"TOOL CALLED: save_memory({content}, {category})") memories = _load_memories() memory_id = str(uuid.uuid4())[:8] memories.append({ "id": memory_id, "content": content, "category": category, "created_at": datetime.datetime.now().isoformat(), }) _save_memories(memories) return f"已保存记忆 [{memory_id}]: {content}" @mcp.tool() def get_memories(category: str = "") -> str: """ 读取长期记忆。这是你的大脑，存储了所有你被要求记住的事情。 什么时候必须调用: - 用户问"你还记得吗"、"你记不记得"、"之前说的"、"上次提到的" - 用户提到之前的对话内容，如"我们要买什么"、"我让你记的那个"、"待办事项" - 用户问"你知道吗"且涉及之前可能保存过的信息 - 每次新对话开始时，先读取记忆了解背景 - 控制设备之前，先读取 device 分类的记忆了解设备经验 - 任何你觉得之前可能保存过相关信息的时候 参数: - category: 可选，按分类过滤。留空返回全部。可选值: device, preference, experience, todo, general """ _log(f"TOOL CALLED: get_memories({category})") memories = _load_memories() if not memories: return "暂无记忆。当你学到重要信息时，请用 save_memory 保存。" if category: memories = [m for m in memories if m.get("category") == category] if not memories: return f"没有分类为'{category}'的记忆" lines = [] for m in memories: cat = m.get("category", "general") lines.append(f"[{m['id']}] ({cat}) {m['content']}") return "\n".join(lines) @mcp.tool() def delete_memory(memory_id: str) -> str: """ 删除一条过时或错误的记忆。 参数: - memory_id: 记忆的ID，通过 get_memories 获取 """ _log(f"TOOL CALLED: delete_memory({memory_id})") memories = _load_memories() new_memories = [m for m in memories if m.get("id") != memory_id] if len(new_memories) == len(memories): return f"未找到ID为 {memory_id} 的记忆" _save_memories(new_memories) return f"已删除记忆 {memory_id}" # ==================== Home Assistant Direct Control ==================== def ha_request(method, path, data=None): url = f"{HA_URL}/api/{path}" _log(f"ha_request {method} {url} data={data}") headers = { "Authorization": f"Bearer {HA_TOKEN}", "Content-Type": "application/json", } body = json.dumps(data).encode() if data else None req = urllib.request.Request(url, data=body, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=10) as resp: result = json.loads(resp.read().decode()) _log(f"ha_request OK, result_type={type(result).__name__}") return result except urllib.error.HTTPError as e: error_body = e.read().decode() if e.fp else "" _log(f"ha_request ERROR: HTTP {e.code}: {error_body[:200]}") return {"error": f"HTTP {e.code}: {error_body}"} except Exception as e: _log(f"ha_request ERROR: {e}") return {"error": str(e)} @mcp.tool() def search_device(keyword: str) -> str: """ 根据关键词搜索智能家居设备。在控制设备之前，必须先用这个工具搜索找到正确的 entity_id。 例如搜索"书房"会返回所有名字包含"书房"的设备。 重要规则： - 如果搜索结果有多个设备匹配，必须询问用户想控制哪一个，不要自己猜测。 - 如果搜索结果只有一个匹配，可以直接控制。 - 如果没有找到匹配设备，告诉用户没有找到，并建议用其他关键词搜索。 参数: - keyword: 搜索关键词，如 书房、客厅、灯、窗帘、空调 """ _log(f"TOOL CALLED: search_device({keyword})") result = ha_request("GET", "states") if isinstance(result, dict) and "error" in result: return f"错误: {result['error']}" matches = [] for entity in result: eid = entity.get("entity_id", "") name = entity.get("attributes", {}).get("friendly_name", eid) state = entity.get("state", "") if keyword.lower() in name.lower() or keyword.lower() in eid.lower(): matches.append(f"{name} ({eid}): {state}") if not matches: return f"未找到包含'{keyword}'的设备" return "\n".join(matches) @mcp.tool() def get_states() -> str: """ 获取所有智能家居设备的状态列表。 返回所有实体的 entity_id、state 和 friendly_name。 用于查看家里有哪些设备以及它们的当前状态。 """ _log("TOOL CALLED: get_states") result = ha_request("GET", "states") if isinstance(result, dict) and "error" in result: return f"错误: {result['error']}" summary = [] for entity in result: eid = entity.get("entity_id", "") state = entity.get("state", "") name = entity.get("attributes", {}).get("friendly_name", eid) summary.append(f"{name} ({eid}): {state}") return "\n".join(summary) @mcp.tool() def get_entity_state(entity_id: str) -> str: """ 获取单个智能家居设备的详细状态。 参数 entity_id: 设备ID，必须通过 search_device 获取真实ID。 """ _log(f"TOOL CALLED: get_entity_state({entity_id})") result = ha_request("GET", f"states/{entity_id}") if isinstance(result, dict) and "error" in result: return f"错误: {result['error']}" state = result.get("state", "unknown") attrs = result.get("attributes", {}) name = attrs.get("friendly_name", entity_id) lines = [f"{name}: {state}"] for k, v in attrs.items(): if k != "friendly_name": lines.append(f" {k}: {v}") return "\n".join(lines) @mcp.tool() def call_service(domain: str, service: str, entity_id: str, extra_data: str = "") -> str: """ 调用 Home Assistant 服务来控制智能家居设备。这是最通用的设备控制工具。 注意：使用前必须先调用 search_device 搜索正确的 entity_id，不要猜测。 参数: - domain: 服务域，如 light, switch, climate, cover, media_player, fan, scene, script, automation - service: 服务名，如 turn_on, turn_off, toggle, open_cover, close_cover, set_cover_position, set_temperature - entity_id: 设备ID，必须通过 search_device 获取真实ID - extra_data: 可选的额外参数JSON字符串，如 {"brightness_pct": 50} 或 {"position": 50} 常用示例: - 开灯: domain="light", service="turn_on", entity_id="light.xxx" - 开开关: domain="switch", service="turn_on", entity_id="switch.xxx" - 开窗帘: domain="cover", service="open_cover", entity_id="cover.xxx" - 关窗帘: domain="cover", service="close_cover", entity_id="cover.xxx" - 执行场景: domain="scene", service="turn_on", entity_id="scene.xxx" """ _log(f"TOOL CALLED: call_service({domain}, {service}, {entity_id}, {extra_data})") payload = {"entity_id": entity_id} if extra_data: try: payload.update(json.loads(extra_data)) except json.JSONDecodeError: return f"错误: extra_data 不是有效的 JSON: {extra_data}" result = ha_request("POST", f"services/{domain}/{service}", payload) if isinstance(result, dict) and "error" in result: return f"错误: {result['error']}" return f"已执行: {domain}.{service} -> {entity_id}" @mcp.tool() def control_light(entity_id: str, action: str, brightness_pct: int = -1, color_temp: int = -1) -> str: """ 控制灯光。用于开灯、关灯、调亮度、调色温。 注意：使用前必须先调用 search_device 搜索正确的 entity_id，不要猜测。 参数: - entity_id: 灯的ID，必须通过 search_device 获取真实ID - action: on/off/toggle - brightness_pct: 亮度百分比 0-100，-1表示不设置 - color_temp: 色温(mireds)，-1表示不设置 """ _log(f"TOOL CALLED: control_light({entity_id}, {action})") data = {"entity_id": entity_id} if brightness_pct >= 0: data["brightness_pct"] = max(0, min(100, brightness_pct)) if color_temp >= 0: data["color_temp"] = color_temp service = {"on": "turn_on", "off": "turn_off", "toggle": "toggle"}.get(action, "turn_on") result = ha_request("POST", f"services/light/{service}", data) if isinstance(result, dict) and "error" in result: return f"错误: {result['error']}" return f"灯光已{'打开' if action == 'on' else '关闭' if action == 'off' else '切换'}: {entity_id}" @mcp.tool() def control_switch(entity_id: str, action: str) -> str: """ 控制开关设备（插座、灯开关、风扇开关等）。很多灯实际上是通过开关(switch)控制的。 注意：使用前必须先调用 search_device 搜索正确的 entity_id，不要猜测。 参数: - entity_id: 开关ID，必须通过 search_device 获取真实ID - action: on/off/toggle """ _log(f"TOOL CALLED: control_switch({entity_id}, {action})") service = {"on": "turn_on", "off": "turn_off", "toggle": "toggle"}.get(action, "toggle") result = ha_request("POST", f"services/switch/{service}", {"entity_id": entity_id}) if isinstance(result, dict) and "error" in result: return f"错误: {result['error']}" return f"开关已{'打开' if action == 'on' else '关闭' if action == 'off' else '切换'}: {entity_id}" @mcp.tool() def control_cover(entity_id: str, action: str, position: int = -1) -> str: """ 控制窗帘/门帘/车库门。 注意：使用前必须先调用 search_device 搜索正确的 entity_id，不要猜测。 参数: - entity_id: 窗帘ID，必须通过 search_device 获取真实ID - action: open/close/stop - position: 位置百分比 0-100(0=全关,100=全开)，-1表示不设置 """ _log(f"TOOL CALLED: control_cover({entity_id}, {action})") if position >= 0: data = {"entity_id": entity_id, "position": max(0, min(100, position))} result = ha_request("POST", "services/cover/set_cover_position", data) else: service = {"open": "open_cover", "close": "close_cover", "stop": "stop_cover"}.get(action, "open_cover") result = ha_request("POST", f"services/cover/{service}", {"entity_id": entity_id}) if isinstance(result, dict) and "error" in result: return f"错误: {result['error']}" return f"窗帘已{'打开' if action == 'open' else '关闭' if action == 'close' else '停止'}: {entity_id}" @mcp.tool() def run_scene(entity_id: str) -> str: """ 执行一个智能家居场景。 注意：使用前必须先调用 search_device 搜索正确的 entity_id，不要猜测。 参数: - entity_id: 场景ID，必须通过 search_device 获取真实ID """ _log(f"TOOL CALLED: run_scene({entity_id})") result = ha_request("POST", "services/scene/turn_on", {"entity_id": entity_id}) if isinstance(result, dict) and "error" in result: return f"错误: {result['error']}" return f"场景已执行: {entity_id}" @mcp.tool() def control_tv(entity_id: str, action: str, volume: int = -1, source: str = "") -> str: """ 控制电视。用于开关电视、调音量、切换信号源。 注意：使用前必须先调用 search_device 搜索正确的 entity_id，不要猜测。 参数: - entity_id: 电视的 media_player ID，必须通过 search_device 获取真实ID - action: on/off/volume_mute/volume_unmute，开关电视或静音 - volume: 音量 0-100，-1表示不设置 - source: 信号源名称，如 HDMI 1, HDMI 2, Netflix 等，空字符串表示不切换 """ _log(f"TOOL CALLED: control_tv({entity_id}, {action}, volume={volume}, source={source})") results = [] if action == "on": results.append(ha_request("POST", "services/media_player/turn_on", {"entity_id": entity_id})) elif action == "off": results.append(ha_request("POST", "services/media_player/turn_off", {"entity_id": entity_id})) elif action == "volume_mute": results.append(ha_request("POST", "services/media_player/volume_mute", {"entity_id": entity_id, "is_volume_muted": True})) elif action == "volume_unmute": results.append(ha_request("POST", "services/media_player/volume_mute", {"entity_id": entity_id, "is_volume_muted": False})) if volume >= 0: vol = max(0, min(100, volume)) / 100.0 results.append(ha_request("POST", "services/media_player/volume_set", {"entity_id": entity_id, "volume_level": vol})) if source: results.append(ha_request("POST", "services/media_player/select_source", {"entity_id": entity_id, "source": source})) for r in results: if isinstance(r, dict) and "error" in r: return f"错误: {r['error']}" parts = [] if action in ("on", "off"): parts.append(f"已{'打开' if action == 'on' else '关闭'}") elif action == "volume_mute": parts.append("已静音") elif action == "volume_unmute": parts.append("已取消静音") if volume >= 0: parts.append(f"音量设为{volume}%") if source: parts.append(f"信号源切换为{source}") return f"电视 {entity_id}: {', '.join(parts)}" if parts else f"电视 {entity_id}: 无操作" @mcp.tool() def control_media(entity_id: str, action: str) -> str: """ 控制媒体播放。适用于所有 media_player 设备（电视、Google Nest Hub、音箱、Xbox 等）。 注意：使用前必须先调用 search_device 搜索正确的 entity_id，不要猜测。 参数: - entity_id: media_player 的ID，必须通过 search_device 获取真实ID - action: play/pause/stop/next/previous - play: 播放 - pause: 暂停 - stop: 停止 - next: 下一曲/下一个 - previous: 上一曲/上一个 """ _log(f"TOOL CALLED: control_media({entity_id}, {action})") service_map = { "play": "media_play", "pause": "media_pause", "stop": "media_stop", "next": "media_next_track", "previous": "media_previous_track", } service = service_map.get(action) if not service: return f"错误: 不支持的操作 '{action}'，可选: play/pause/stop/next/previous" result = ha_request("POST", f"services/media_player/{service}", {"entity_id": entity_id}) if isinstance(result, dict) and "error" in result: return f"错误: {result['error']}" action_names = {"play": "播放", "pause": "暂停", "stop": "停止", "next": "下一曲", "previous": "上一曲"} return f"已{action_names.get(action, action)}: {entity_id}" @mcp.tool() def play_on_device(entity_id: str, media_url: str, media_type: str = "youtube") -> str: """ 在指定设备上播放内容。可以在 Google Nest Hub 上播放视频，在音箱上播放音乐等。 重要：YouTube 视频只能通过 Google Cast 协议播放。设备必须是 Cast 实体，例如客厅display、客厅Google、電視(media_player.dian_shi)。注意客厅电视有两个实体：media_player.dian_shi 是 Cast 实体（能播YouTube），media_player.dian_shi_2 是 Bravia 实体（不能播YouTube）。 注意：使用前必须先调用 search_device 搜索正确的 entity_id，不要猜测。 参数: - entity_id: media_player 的ID，必须通过 search_device 获取真实ID。播放 YouTube 必须用 Cast 实体。 - media_url: 媒体URL或内容ID - YouTube 视频: 填 YouTube URL，如 https://www.youtube.com/watch?v=xxxxx - 音乐: 填音乐文件URL - 网络电台: 填电台流URL - media_type: 媒体类型，可选 youtube/music/video/url，默认 youtube """ _log(f"TOOL CALLED: play_on_device({entity_id}, {media_url}, {media_type})") if media_type == "youtube" or "youtube.com" in media_url or "youtu.be" in media_url: video_id = media_url if "youtube.com" in media_url and "v=" in media_url: video_id = media_url.split("v=")[1].split("&")[0] elif "youtu.be/" in media_url: video_id = media_url.split("youtu.be/")[1].split("?")[0] data = { "entity_id": entity_id, "media_content_type": "cast", "media_content_id": json.dumps({ "app_name": "youtube", "media_id": video_id, }), } else: data = { "entity_id": entity_id, "media_content_id": media_url, "media_content_type": media_type, } result = ha_request("POST", "services/media_player/play_media", data) if isinstance(result, dict) and "error" in result: err = result["error"] return f"错误: {err}" @mcp.tool() def send_tts(entity_id: str, message: str) -> str: """ 让指定设备说一段话（文字转语音）。可以让 Google Nest Hub 或 Google 音箱播报消息。 注意：使用前必须先调用 search_device 搜索正确的 entity_id，不要猜测。 参数: - entity_id: media_player 的ID（Google Nest Hub 或音箱），必须通过 search_device 获取真实ID - message: 要播报的文字内容，如 "该吃饭了"、"有人按门铃" """ _log(f"TOOL CALLED: send_tts({entity_id}, {message})") data = { "entity_id": HA_TTS_ENGINE, "media_player_entity_id": entity_id, "message": message, } result = ha_request("POST", "services/tts/speak", data) if isinstance(result, dict) and "error" in result: err = result["error"] return f"错误: {err}" return f"已播报: {message}" @mcp.tool() def send_remote_command(entity_id: str, command: str) -> str: """ 向电视发送遥控器按键指令。 注意：使用前必须先调用 search_device 搜索正确的 entity_id（remote.xxx），不要猜测。 参数: - entity_id: 遥控器的 remote ID（如 remote.dian_shi），必须通过 search_device 获取 - command: 遥控按键名称，常用按键: - 导航: Up, Down, Left, Right, Confirm/Select, Back, Home - 媒体: Play, Pause, Stop, Next, Previous - 音量: VolumeUp, VolumeDown, Mute - 电源: PowerOff - 数字: Num0-Num9 - 其他: Input, Guide, EPG, ChannelUp, ChannelDown """ _log(f"TOOL CALLED: send_remote_command({entity_id}, {command})") data = { "entity_id": entity_id, "command": command, } result = ha_request("POST", "services/remote/send_command", data) if isinstance(result, dict) and "error" in result: return f"错误: {result['error']}" return f"已发送遥控指令 '{command}' -> {entity_id}" # ==================== Google Services ==================== GOOGLE_CLIENT_ID = os.getenv("GOOGLE_CLIENT_ID", "").strip() GOOGLE_CLIENT_SECRET = os.getenv("GOOGLE_CLIENT_SECRET", "").strip() GOOGLE_REFRESH_TOKEN = os.getenv("GOOGLE_REFRESH_TOKEN", "").strip() GOOGLE_MAPS_API_KEY = os.getenv("GOOGLE_MAPS_API_KEY", "").strip() _google_access_token = {"token": "", "expires": 0} def _google_get_token(): """Get or refresh Google OAuth2 access token.""" import time now = time.time() if _google_access_token["token"] and now < _google_access_token["expires"] - 60: return _google_access_token["token"] data = urllib.parse.urlencode({ "client_id": GOOGLE_CLIENT_ID, "client_secret": GOOGLE_CLIENT_SECRET, "refresh_token": GOOGLE_REFRESH_TOKEN, "grant_type": "refresh_token", }).encode() req = urllib.request.Request("https://oauth2.googleapis.com/token", data=data) with urllib.request.urlopen(req, timeout=10) as resp: result = json.loads(resp.read().decode()) _google_access_token["token"] = result["access_token"] _google_access_token["expires"] = now + result.get("expires_in", 3600) return result["access_token"] def _google_api(method, url, body=None): """Call Google API with auto token refresh.""" _log(f"Google API: {method} {url[:100]}") token = _google_get_token() headers = {"Authorization": f"Bearer {token}"} data = None if body is not None: data = json.dumps(body).encode() headers["Content-Type"] = "application/json" req = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=15) as resp: return json.loads(resp.read().decode()) except urllib.error.HTTPError as e: err_body = e.read().decode()[:300] _log(f"Google API error: {e.code} {err_body}") return {"error": f"HTTP {e.code}: {err_body}"} except Exception as e: _log(f"Google API exception: {e}") return {"error": str(e)} # ---------- Gmail ---------- @mcp.tool() def gmail_list(query: str = "is:unread", max_results: int = 10) -> str: """ 查看邮件列表。 什么时候调用: - 用户说"看看邮件"、"有没有新邮件"、"查一下邮箱"、"最近的邮件" - 用户问某人有没有发邮件来 参数: - query: Gmail搜索语法。常用: "is:unread"(未读), "from:xxx"(来自某人), "subject:xxx"(主题含), "newer_than:1d"(最近1天), "is:important" - max_results: 返回数量，默认10 """ _log(f"TOOL CALLED: gmail_list({query}, {max_results})") qs = urllib.parse.urlencode({"q": query, "maxResults": min(max_results, 20)}) data = _google_api("GET", f"https://gmail.googleapis.com/gmail/v1/users/me/messages?{qs}") if "error" in data: return f"查询失败: {data['error']}" messages = data.get("messages", []) if not messages: return f"没有符合条件的邮件 (搜索: {query})" results = [] for msg in messages[:max_results]: detail = _google_api("GET", f"https://gmail.googleapis.com/gmail/v1/users/me/messages/{msg['id']}?format=metadata&metadataHeaders=From&metadataHeaders=Subject&metadataHeaders=Date") if "error" in detail: continue headers = {h["name"]: h["value"] for h in detail.get("payload", {}).get("headers", [])} subj = headers.get("Subject", "(无主题)") frm = headers.get("From", "未知") date = headers.get("Date", "")[:25] snippet = detail.get("snippet", "")[:80] labels = detail.get("labelIds", []) unread = "📩" if "UNREAD" in labels else " " results.append(f"{unread} {date}\n 从: {frm}\n 主题: {subj}\n 预览: {snippet}\n ID: {msg['id']}") return f"找到 {len(messages)} 封邮件:\n\n" + "\n\n".join(results) @mcp.tool() def gmail_read(message_id: str) -> str: """ 读取一封邮件的完整内容。需要先用 gmail_list 获取邮件ID。 参数: - message_id: 邮件ID，从 gmail_list 结果中获取 """ _log(f"TOOL CALLED: gmail_read({message_id})") data = _google_api("GET", f"https://gmail.googleapis.com/gmail/v1/users/me/messages/{message_id}?format=full") if "error" in data: return f"读取失败: {data['error']}" headers = {h["name"]: h["value"] for h in data.get("payload", {}).get("headers", [])} subj = headers.get("Subject", "(无主题)") frm = headers.get("From", "未知") to = headers.get("To", "未知") date = headers.get("Date", "") import base64 body = "" payload = data.get("payload", {}) if payload.get("body", {}).get("data"): body = base64.urlsafe_b64decode(payload["body"]["data"]).decode("utf-8", errors="replace") elif payload.get("parts"): for part in payload["parts"]: if part.get("mimeType") == "text/plain" and part.get("body", {}).get("data"): body = base64.urlsafe_b64decode(part["body"]["data"]).decode("utf-8", errors="replace") break if not body: for part in payload["parts"]: if part.get("mimeType") == "text/html" and part.get("body", {}).get("data"): raw = base64.urlsafe_b64decode(part["body"]["data"]).decode("utf-8", errors="replace") import re body = re.sub(r"<[^>]+>", "", raw)[:2000] break if not body: body = data.get("snippet", "(无法提取正文)") return f"日期: {date}\n从: {frm}\n到: {to}\n主题: {subj}\n\n{body[:3000]}" @mcp.tool() def gmail_send(to: str, subject: str, body: str) -> str: """ 发送邮件。 什么时候调用: - 用户说"发邮件给xxx"、"帮我写封邮件"、"回复那封邮件" 参数: - to: 收件人邮箱 - subject: 邮件主题 - body: 邮件正文 """ _log(f"TOOL CALLED: gmail_send({to}, {subject})") import base64 message = f"To: {to}\r\nSubject: {subject}\r\nContent-Type: text/plain; charset=utf-8\r\n\r\n{body}" raw = base64.urlsafe_b64encode(message.encode("utf-8")).decode("ascii") data = _google_api("POST", "https://gmail.googleapis.com/gmail/v1/users/me/messages/send", {"raw": raw}) if "error" in data: return f"发送失败: {data['error']}" return f"邮件已发送给 {to}，主题: {subject}" # ---------- Google Calendar ---------- @mcp.tool() def calendar_list(days: int = 7) -> str: """ 查看日程安排。 什么时候调用: - 用户说"今天有什么安排"、"这周日程"、"看看日历"、"明天有什么事" - 用户问某天有没有空 参数: - days: 查看未来几天的日程，默认7天 """ _log(f"TOOL CALLED: calendar_list({days})") now = datetime.datetime.utcnow() time_min = now.strftime("%Y-%m-%dT%H:%M:%SZ") time_max = (now + datetime.timedelta(days=days)).strftime("%Y-%m-%dT%H:%M:%SZ") qs = urllib.parse.urlencode({ "timeMin": time_min, "timeMax": time_max, "maxResults": 20, "singleEvents": "true", "orderBy": "startTime", }) data = _google_api("GET", f"https://www.googleapis.com/calendar/v3/calendars/primary/events?{qs}") if "error" in data: return f"查询失败: {data['error']}" events = data.get("items", []) if not events: return f"未来 {days} 天没有日程安排" results = [] for e in events: start = e.get("start", {}).get("dateTime", e.get("start", {}).get("date", "")) end = e.get("end", {}).get("dateTime", e.get("end", {}).get("date", "")) summary = e.get("summary", "(无标题)") location = e.get("location", "") loc_str = f"\n 地点: {location}" if location else "" description = e.get("description", "") desc_str = f"\n 备注: {description[:100]}" if description else "" results.append(f"📅 {start[:16]} ~ {end[11:16]}\n {summary}{loc_str}{desc_str}") return f"未来 {days} 天有 {len(events)} 个日程:\n\n" + "\n\n".join(results) @mcp.tool() def calendar_create(summary: str, start_time: str, end_time: str = "", description: str = "", location: str = "") -> str: """ 创建日程。 什么时候调用: - 用户说"帮我加个日程"、"安排一下"、"提醒我"、"约个会"、"记到日历里" 参数: - summary: 日程标题，如 "开会"、"看医生" - start_time: 开始时间，格式 "2026-06-16T14:00:00"（包含日期和时间）或 "2026-06-16"（全天事件） - end_time: 结束时间，同上格式。不填则默认1小时后 - description: 备注信息 - location: 地点 """ _log(f"TOOL CALLED: calendar_create({summary}, {start_time})") event = {"summary": summary} if description: event["description"] = description if location: event["location"] = location if "T" in start_time: event["start"] = {"dateTime": start_time, "timeZone": "Pacific/Auckland"} if not end_time: h = int(start_time[11:13]) end_time = start_time[:11] + f"{h+1:02d}" + start_time[13:] event["end"] = {"dateTime": end_time, "timeZone": "Pacific/Auckland"} else: event["start"] = {"date": start_time} event["end"] = {"date": end_time or start_time} data = _google_api("POST", "https://www.googleapis.com/calendar/v3/calendars/primary/events", event) if "error" in data: return f"创建失败: {data['error']}" return f"日程已创建: {summary}\n时间: {start_time}" # ---------- Google Tasks ---------- @mcp.tool() def tasks_list(show_completed: bool = False) -> str: """ 查看待办事项列表。 什么时候调用: - 用户说"待办事项"、"我的任务"、"要做什么"、"todo" - 用户问"还有什么没做" 参数: - show_completed: 是否显示已完成的任务，默认不显示 """ _log(f"TOOL CALLED: tasks_list({show_completed})") lists_data = _google_api("GET", "https://tasks.googleapis.com/tasks/v1/users/@me/lists") if "error" in lists_data: return f"查询失败: {lists_data['error']}" all_tasks = [] for tl in lists_data.get("items", []): list_id = tl["id"] list_title = tl.get("title", "默认") qs = "showCompleted=true&showHidden=true" if show_completed else "showCompleted=false" tasks_data = _google_api("GET", f"https://tasks.googleapis.com/tasks/v1/lists/{list_id}/tasks?{qs}&maxResults=50") if "error" in tasks_data: continue for t in tasks_data.get("items", []): status = "✅" if t.get("status") == "completed" else "⬜" title = t.get("title", "") if not title: continue due = t.get("due", "")[:10] due_str = f" (截止: {due})" if due else "" notes = t.get("notes", "") notes_str = f"\n 备注: {notes[:80]}" if notes else "" all_tasks.append(f" {status} {title}{due_str}{notes_str}\n 列表: {list_title} | ID: {t['id']}") if not all_tasks: return "没有待办事项" return f"待办事项 ({len(all_tasks)} 项):\n\n" + "\n\n".join(all_tasks) @mcp.tool() def tasks_add(title: str, notes: str = "", due_date: str = "") -> str: """ 添加待办事项。 什么时候调用: - 用户说"帮我加个待办"、"添加任务"、"记一下要做的事" - 用户说"提醒我做xxx" 参数: - title: 任务标题 - notes: 备注详情 - due_date: 截止日期，格式 "2026-06-20"，可不填 """ _log(f"TOOL CALLED: tasks_add({title})") lists_data = _google_api("GET", "https://tasks.googleapis.com/tasks/v1/users/@me/lists") if "error" in lists_data: return f"失败: {lists_data['error']}" list_id = lists_data["items"][0]["id"] task = {"title": title} if notes: task["notes"] = notes if due_date: task["due"] = f"{due_date}T00:00:00.000Z" data = _google_api("POST", f"https://tasks.googleapis.com/tasks/v1/lists/{list_id}/tasks", task) if "error" in data: return f"添加失败: {data['error']}" return f"已添加待办: {title}" @mcp.tool() def tasks_complete(task_id: str) -> str: """ 完成一个待办事项。需要先用 tasks_list 获取任务ID。 参数: - task_id: 任务ID """ _log(f"TOOL CALLED: tasks_complete({task_id})") lists_data = _google_api("GET", "https://tasks.googleapis.com/tasks/v1/users/@me/lists") if "error" in lists_data: return f"失败: {lists_data['error']}" for tl in lists_data.get("items", []): list_id = tl["id"] data = _google_api("PATCH", f"https://tasks.googleapis.com/tasks/v1/lists/{list_id}/tasks/{task_id}", {"status": "completed"}) if "error" not in data: return f"已完成任务: {data.get('title', task_id)}" return f"未找到任务 {task_id}" # ---------- Google Drive ---------- @mcp.tool() def drive_search(query: str = "", max_results: int = 10) -> str: """ 搜索 Google Drive 文件。 什么时候调用: - 用户说"找一下那个文件"、"我的文档"、"最近的文件" 参数: - query: 搜索关键词，留空返回最近文件 - max_results: 返回数量 """ _log(f"TOOL CALLED: drive_search({query})") params = {"pageSize": min(max_results, 20), "fields": "files(id,name,mimeType,modifiedTime,size,webViewLink)", "orderBy": "modifiedTime desc"} if query: params["q"] = f"name contains '{query}'" qs = urllib.parse.urlencode(params) data = _google_api("GET", f"https://www.googleapis.com/drive/v3/files?{qs}") if "error" in data: return f"搜索失败: {data['error']}" files = data.get("files", []) if not files: return f"未找到相关文件" results = [] for f in files: modified = f.get("modifiedTime", "")[:10] name = f.get("name", "") link = f.get("webViewLink", "") mime = f.get("mimeType", "").split(".")[-1] results.append(f"📄 {name}\n 类型: {mime} | 修改: {modified}\n 链接: {link}") return "\n\n".join(results) # ---------- Google Contacts ---------- @mcp.tool() def contacts_search(name: str) -> str: """ 搜索通讯录中的联系人。 什么时候调用: - 用户说"xxx的电话号码"、"xxx的邮箱"、"查一下xxx的联系方式" 参数: - name: 联系人姓名 """ _log(f"TOOL CALLED: contacts_search({name})") qs = urllib.parse.urlencode({"query": name, "readMask": "names,emailAddresses,phoneNumbers", "pageSize": 10}) data = _google_api("GET", f"https://people.googleapis.com/v1/people:searchContacts?{qs}") if "error" in data: return f"搜索失败: {data['error']}" results_list = data.get("results", []) if not results_list: return f"未找到名为'{name}'的联系人" results = [] for r in results_list: person = r.get("person", {}) pname = person.get("names", [{}])[0].get("displayName", "未知") phones = [p.get("value", "") for p in person.get("phoneNumbers", [])] emails = [e.get("value", "") for e in person.get("emailAddresses", [])] phone_str = ", ".join(phones) if phones else "无" email_str = ", ".join(emails) if emails else "无" results.append(f"👤 {pname}\n 电话: {phone_str}\n 邮箱: {email_str}") return "\n\n".join(results) # ---------- Google Maps ---------- @mcp.tool() def maps_directions(origin: str, destination: str, mode: str = "driving") -> str: """ 查询路线和导航信息。 什么时候调用: - 用户说"从A到B怎么走"、"多远"、"多久能到"、"路线" 参数: - origin: 出发地，如 "家"、"奥克兰"、具体地址 - destination: 目的地 - mode: 交通方式，driving(开车)、walking(步行)、transit(公交)、bicycling(骑车) """ _log(f"TOOL CALLED: maps_directions({origin}, {destination}, {mode})") qs = urllib.parse.urlencode({ "origin": origin, "destination": destination, "mode": mode, "key": GOOGLE_MAPS_API_KEY, "language": "zh-CN", }) url = f"https://maps.googleapis.com/maps/api/directions/json?{qs}" try: req = urllib.request.Request(url) with urllib.request.urlopen(req, timeout=10) as resp: data = json.loads(resp.read().decode()) except Exception as e: return f"查询失败: {e}" if data.get("status") != "OK": return f"未找到路线: {data.get('status')}" leg = data["routes"][0]["legs"][0] distance = leg["distance"]["text"] duration = leg["duration"]["text"] start = leg["start_address"] end = leg["end_address"] mode_cn = {"driving": "驾车", "walking": "步行", "transit": "公交", "bicycling": "骑车"}.get(mode, mode) steps = [] for i, s in enumerate(leg["steps"][:10], 1): import re instruction = re.sub(r"<[^>]+>", "", s.get("html_instructions", "")) steps.append(f" {i}. {instruction} ({s['distance']['text']})") return f"🗺️ {mode_cn}路线:\n从: {start}\n到: {end}\n距离: {distance}\n时间: {duration}\n\n路线步骤:\n" + "\n".join(steps) @mcp.tool() def maps_nearby(keyword: str, location: str = "", radius: int = 2000) -> str: """ 搜索附近的商家/地点。 什么时候调用: - 用户说"附近有什么餐厅"、"最近的加油站"、"附近的超市" 参数: - keyword: 搜索关键词，如 "餐厅"、"加油站"、"咖啡" - location: 搜索中心位置，如 "奥克兰市中心"。留空则需要用户指定 - radius: 搜索半径(米)，默认2000 """ _log(f"TOOL CALLED: maps_nearby({keyword}, {location}, {radius})") if not location: return "请告诉我你想搜索哪个位置附近的" + keyword # First geocode the location geo_qs = urllib.parse.urlencode({"address": location, "key": GOOGLE_MAPS_API_KEY}) try: req = urllib.request.Request(f"https://maps.googleapis.com/maps/api/geocode/json?{geo_qs}") with urllib.request.urlopen(req, timeout=10) as resp: geo_data = json.loads(resp.read().decode()) except Exception as e: return f"地址解析失败: {e}" if not geo_data.get("results"): return f"无法找到位置: {location}" loc = geo_data["results"][0]["geometry"]["location"] lat, lng = loc["lat"], loc["lng"] # Search nearby qs = urllib.parse.urlencode({ "location": f"{lat},{lng}", "radius": radius, "keyword": keyword, "key": GOOGLE_MAPS_API_KEY, "language": "zh-CN", }) try: req = urllib.request.Request(f"https://maps.googleapis.com/maps/api/place/nearbysearch/json?{qs}") with urllib.request.urlopen(req, timeout=10) as resp: data = json.loads(resp.read().decode()) except Exception as e: return f"搜索失败: {e}" places = data.get("results", []) if not places: return f"在{location}附近未找到{keyword}" results = [] for p in places[:8]: name = p.get("name", "") addr = p.get("vicinity", "") rating = p.get("rating", "N/A") total = p.get("user_ratings_total", 0) is_open = p.get("opening_hours", {}).get("open_now") open_str = " 🟢营业中" if is_open else (" 🔴已关门" if is_open is False else "") results.append(f"📍 {name}{open_str}\n 地址: {addr}\n 评分: {rating} ({total}条评价)") return f"{location}附近的{keyword} (找到{len(places)}个):\n\n" + "\n\n".join(results) # ==================== YouTube ==================== def _youtube_api(path, params): """Call YouTube Data API v3.""" params["key"] = YOUTUBE_API_KEY qs = urllib.parse.urlencode(params) url = f"https://www.googleapis.com/youtube/v3/{path}?{qs}" _log(f"YouTube API: {path} params={params}") req = urllib.request.Request(url) try: with urllib.request.urlopen(req, timeout=10) as resp: return json.loads(resp.read().decode()) except Exception as e: _log(f"YouTube API error: {e}") return {"error": str(e)} @mcp.tool() def youtube_search(keyword: str, max_results: int = 5) -> str: """ 搜索 YouTube 视频。返回视频标题、频道名和视频ID。 参数: - keyword: 搜索关键词，如 "我爱纽西兰"、"周杰伦 晴天"、"lofi music" - max_results: 返回结果数量，默认5，最多10 """ _log(f"TOOL CALLED: youtube_search({keyword}, {max_results})") if not YOUTUBE_API_KEY: return "错误: 未配置 YOUTUBE_API_KEY" max_results = min(max(1, max_results), 10) data = _youtube_api("search", { "part": "snippet", "q": keyword, "maxResults": max_results, "type": "video", }) if "error" in data: return f"搜索失败: {data['error']}" items = data.get("items", []) if not items: return f"未找到与'{keyword}'相关的视频" results = [] for i, item in enumerate(items, 1): vid = item["id"]["videoId"] title = item["snippet"]["title"] channel = item["snippet"]["channelTitle"] url = f"https://www.youtube.com/watch?v={vid}" results.append(f"{i}. {title}\n 频道: {channel}\n 链接: {url}") return "\n\n".join(results) @mcp.tool() def youtube_play(keyword: str, device_entity_id: str) -> str: """ 搜索 YouTube 视频并在指定设备上播放第一个结果。一步到位的快捷工具。 重要：只能通过 Google Cast 协议播放。要投到电视请用 Cast 实体（如 media_player.dian_shi 電視），不要用 Bravia 实体（media_player.dian_shi_2）。 注意：device_entity_id 必须先通过 search_device 获取真实ID，不要猜测。 参数: - keyword: 搜索关键词，如 "我爱纽西兰 最新"、"周杰伦 晴天 MV" - device_entity_id: Google Cast 设备的 media_player ID（如 客厅display、media_player.dian_shi 電視） """ _log(f"TOOL CALLED: youtube_play({keyword}, {device_entity_id})") if not YOUTUBE_API_KEY: return "错误: 未配置 YOUTUBE_API_KEY" data = _youtube_api("search", { "part": "snippet", "q": keyword, "maxResults": 1, "type": "video", }) if "error" in data: return f"搜索失败: {data['error']}" items = data.get("items", []) if not items: return f"未找到与'{keyword}'相关的视频" vid = items[0]["id"]["videoId"] title = items[0]["snippet"]["title"] video_url = f"https://www.youtube.com/watch?v={vid}" play_data = { "entity_id": device_entity_id, "media_content_type": "cast", "media_content_id": json.dumps({ "app_name": "youtube", "media_id": vid, }), } result = ha_request("POST", "services/media_player/play_media", play_data) if isinstance(result, dict) and "error" in result: err = result["error"] return f"搜索到了但播放失败: {err}\n视频: {title}\n链接: {video_url}" return f"正在播放: {title}\n链接: {video_url}" @mcp.tool() def youtube_get_channel_latest(channel_name: str, device_entity_id: str = "") -> str: """ 获取指定 YouTube 频道的最新视频，可选择直接在设备上播放。 参数: - channel_name: 频道名称或关键词，如 "我爱纽西兰"、"老高與小茉" - device_entity_id: 可选，Google Cast 设备的 media_player ID（如 客厅display、media_player.dian_shi 電視）。如果提供则自动播放最新视频，不提供则只返回视频列表。 """ _log(f"TOOL CALLED: youtube_get_channel_latest({channel_name}, {device_entity_id})") if not YOUTUBE_API_KEY: return "错误: 未配置 YOUTUBE_API_KEY" data = _youtube_api("search", { "part": "snippet", "q": channel_name, "maxResults": 5, "type": "video", "order": "date", }) if "error" in data: return f"搜索失败: {data['error']}" items = data.get("items", []) if not items: return f"未找到'{channel_name}'的相关视频" results = [] for i, item in enumerate(items, 1): vid = item["id"]["videoId"] title = item["snippet"]["title"] channel = item["snippet"]["channelTitle"] published = item["snippet"]["publishedAt"][:10] url = f"https://www.youtube.com/watch?v={vid}" results.append(f"{i}. [{published}] {title}\n 频道: {channel}\n 链接: {url}") if device_entity_id: first_vid = items[0]["id"]["videoId"] first_title = items[0]["snippet"]["title"] first_url = f"https://www.youtube.com/watch?v={first_vid}" play_data = { "entity_id": device_entity_id, "media_content_type": "cast", "media_content_id": json.dumps({ "app_name": "youtube", "media_id": first_vid, }), } play_result = ha_request("POST", "services/media_player/play_media", play_data) if isinstance(play_result, dict) and "error" in play_result: err = play_result["error"] return f"找到视频但播放失败: {err}\n\n" + "\n\n".join(results) return f"正在播放最新视频: {first_title}\n\n其他最新视频:\n" + "\n\n".join(results[1:]) return "\n\n".join(results) # ==================== Vision ==================== def _minimax_vision(image_b64: str, question: str, mime: str = "image/jpeg") -> str: """Call MiniMax-M3 to analyze a base64-encoded image.""" if not MINIMAX_API_KEY: return "错误: 未配置 MINIMAX_API_KEY" import base64 as _b64 image_content = {"type": "image_url", "image_url": {"url": f"data:{mime};base64,{image_b64}"}} payload = json.dumps({ "model": "MiniMax-M3", "messages": [{"role": "user", "content": [image_content, {"type": "text", "text": question}]}], "max_tokens": 1024, }).encode() req = urllib.request.Request( "https://api.minimax.chat/v1/chat/completions", data=payload, headers={"Authorization": f"Bearer {MINIMAX_API_KEY}", "Content-Type": "application/json"}, method="POST", ) try: with urllib.request.urlopen(req, timeout=30) as resp: result = json.loads(resp.read().decode()) return result["choices"][0]["message"]["content"] except urllib.error.HTTPError as e: err = e.read().decode()[:300] _log(f"MiniMax vision error: {e.code} {err}") return f"视觉分析失败: HTTP {e.code}: {err}" except Exception as e: _log(f"MiniMax vision error: {e}") return f"视觉分析失败: {e}" def _fetch_image_b64(url: str): """Download image from URL, return (base64_str, mime_type).""" import base64 as _b64 req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"}) with urllib.request.urlopen(req, timeout=15) as resp: content_type = resp.headers.get("Content-Type", "image/jpeg").split(";")[0].strip() data = resp.read() mime = content_type if content_type.startswith("image/") else "image/jpeg" return _b64.b64encode(data).decode(), mime @mcp.tool() def analyze_image(image_url: str, question: str = "请详细描述这张图片里有什么") -> str: """ 分析一张图片。用 MiniMax 视觉模型理解图片内容。 什么时候调用: - 用户发来图片链接，问"这是什么"、"帮我看看"、"图片里有什么" - 用户说"帮我识别一下这个"并附上图片URL - 用户问图片里的文字、物品、人物、场景 参数: - image_url: 图片的网络链接（http/https URL） - question: 对图片要问的问题，默认描述图片内容 """ _log(f"TOOL CALLED: analyze_image({image_url[:80]}, {question})") try: b64, mime = _fetch_image_b64(image_url) except Exception as e: return f"下载图片失败: {e}" return _minimax_vision(b64, question, mime) async def _get_camera_hls_url(entity_id: str) -> str: """Get a fresh HLS stream URL from HA via websocket.""" ws_url = HA_URL.replace("http://", "ws://").replace("https://", "wss://") + "/api/websocket" async with websockets.connect(ws_url) as ws: msg = json.loads(await asyncio.wait_for(ws.recv(), timeout=10)) if msg.get("type") == "auth_required": await ws.send(json.dumps({"type": "auth", "access_token": HA_TOKEN})) msg = json.loads(await asyncio.wait_for(ws.recv(), timeout=10)) if msg.get("type") != "auth_ok": raise ConnectionError(f"HA auth failed: {msg}") await ws.send(json.dumps({"id": 1, "type": "camera/stream", "entity_id": entity_id})) msg = json.loads(await asyncio.wait_for(ws.recv(), timeout=15)) if not msg.get("success"): raise RuntimeError(f"HA stream error: {msg}") return HA_URL + msg["result"]["url"] @mcp.tool() async def view_camera(entity_id: str = "camera.127_0_0_1", question: str = "请描述摄像头画面里有什么") -> str: """ 查看家里的摄像头画面并用AI描述内容。 什么时候调用: - 用户说"看看摄像头"、"门口有没有人"、"外面什么情况" - 用户说"帮我看看xxx摄像头" - 用户问家里当前的情况 参数: - entity_id: 摄像头的 HA entity_id，默认 camera.127_0_0_1。可用 search_device 搜索其他摄像头。 - question: 要问的问题，默认描述画面内容 """ _log(f"TOOL CALLED: view_camera({entity_id}, {question})") import base64, subprocess, tempfile, shutil if not shutil.which("ffmpeg"): return "错误: 服务器未安装 ffmpeg，无法抓取摄像头画面" try: hls_url = await _get_camera_hls_url(entity_id) except Exception as e: _log(f"view_camera get_hls error: {e}") return f"获取摄像头流地址失败: {e}" with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as tmp: out_path = tmp.name try: result = subprocess.run([ "ffmpeg", "-y", "-headers", f"Authorization: Bearer {HA_TOKEN}\r\n", "-i", hls_url, "-vframes", "1", "-q:v", "2", out_path, ], capture_output=True, text=True, timeout=30) if not os.path.exists(out_path) or os.path.getsize(out_path) == 0: _log(f"ffmpeg failed: {result.stderr[-300:]}") return f"摄像头抓帧失败: {result.stderr[-200:]}" with open(out_path, "rb") as f: image_b64 = base64.b64encode(f.read()).decode() finally: if os.path.exists(out_path): os.unlink(out_path) return _minimax_vision(image_b64, question, "image/jpeg") # ==================== Web Search ==================== @mcp.tool() def web_search(query: str, max_results: int = 5) -> str: """ 用浏览器搜索互联网上的信息。适合回答实时问题、新闻、天气、价格、知识查询等。 什么时候调用: - 用户说"搜一下"、"查一查"、"帮我搜索"、"上网查" - 用户问实时信息：今天的新闻、天气、比赛结果、股价等 - 用户问你不确定的知识点 - 用户问最新的信息 参数: - query: 搜索关键词，越具体越好 - max_results: 返回结果数量，默认5，最多10 """ _log(f"TOOL CALLED: web_search({query}, {max_results})") try: from ddgs import DDGS max_results = min(max(1, max_results), 10) with DDGS() as ddgs: results = list(ddgs.text(query, max_results=max_results)) if not results: return f"没有找到关于「{query}」的搜索结果" lines = [f"搜索「{query}」的结果:\n"] for i, r in enumerate(results, 1): title = r.get("title", "") body = r.get("body", "") href = r.get("href", "") lines.append(f"{i}. {title}\n {body[:200]}\n 链接: {href}") return "\n\n".join(lines) except Exception as e: _log(f"web_search error: {e}") return f"搜索出错: {e}" @mcp.tool() def web_fetch(url: str) -> str: """ 读取一个网页的文字内容。适合用户想看某个具体网页的内容。 参数: - url: 要读取的网页链接 """ _log(f"TOOL CALLED: web_fetch({url})") try: import re headers = {"User-Agent": "Mozilla/5.0 (compatible; XiaozhiBot/1.0)"} req = urllib.request.Request(url, headers=headers) with urllib.request.urlopen(req, timeout=15) as resp: raw = resp.read().decode("utf-8", errors="replace") text = re.sub(r"]*>.*?", "", raw, flags=re.DOTALL) text = re.sub(r"]*>.*?", "", text, flags=re.DOTALL) text = re.sub(r"<[^>]+>", "", text) text = re.sub(r"\s{3,}", "\n\n", text) return text[:3000] except Exception as e: _log(f"web_fetch error: {e}") return f"读取网页出错: {e}" # ==================== OpenClaw (for non-device tasks) ==================== def extract_text(msg): if msg.get("text"): return msg["text"] content = msg.get("content") if isinstance(content, str): return content if isinstance(content, list): parts = [] for item in content: if isinstance(item, dict) and item.get("type") == "text" and item.get("text"): parts.append(item["text"]) elif isinstance(item, str): parts.append(item) return "\n".join(parts) if parts else "" return "" async def ws_connect(): headers = {} if OPENCLAW_TOKEN: headers["Authorization"] = f"Bearer {OPENCLAW_TOKEN}" headers["Origin"] = OPENCLAW_HTTP try: return websockets.connect( OPENCLAW_WS_URL, additional_headers=headers, ping_interval=20, ping_timeout=20, ) except TypeError: return websockets.connect( OPENCLAW_WS_URL, extra_headers=headers, ping_interval=20, ping_timeout=20, ) async def recv_json(ws, timeout=30): raw = await asyncio.wait_for(ws.recv(), timeout=timeout) logger.info("WS recv: %s", raw[:800]) return json.loads(raw) async def send_frame(ws, frame): logger.info("WS send: %s", json.dumps(frame, ensure_ascii=False)[:500]) await ws.send(json.dumps(frame, ensure_ascii=False)) async def send_req(ws, method, params, timeout=30): req_id = str(uuid.uuid4()) await send_frame(ws, {"type": "req", "id": req_id, "method": method, "params": params}) while True: data = await recv_json(ws, timeout=timeout) if data.get("type") == "res" and data.get("id") == req_id: return data async def handshake(ws): first = await recv_json(ws, timeout=10) if first.get("type") == "event" and first.get("event") == "connect.challenge": logger.info("Got connect.challenge") else: logger.warning("Unexpected first message: %s", first.get("type")) res = await send_req(ws, "connect", { "minProtocol": 3, "maxProtocol": 4, "client": { "id": "openclaw-control-ui", "version": "1.0.0", "platform": "linux", "mode": "webchat", }, "role": "operator", "scopes": ["operator.read", "operator.write"], "caps": [], "commands": [], "permissions": {}, "auth": {"token": OPENCLAW_TOKEN}, "locale": "zh-CN", "userAgent": "xiaozhi-mcp/1.0.0", }, timeout=15) if not res or not res.get("ok"): err = res.get("error", res) if res else "no response" raise ConnectionError(f"Handshake failed: {err}") logger.info("Connected to openclaw") return res async def send_chat_and_wait(user_message, session_key="main", timeout=180): async with await ws_connect() as ws: await handshake(ws) sub_res = await send_req(ws, "sessions.messages.subscribe", {"key": session_key}, timeout=10) logger.info("Subscribe result: ok=%s", sub_res.get("ok") if sub_res else None) req_id = str(uuid.uuid4()) await send_frame(ws, { "type": "req", "id": req_id, "method": "chat.send", "params": { "sessionKey": session_key, "message": user_message, "deliver": False, "idempotencyKey": req_id, }, }) collected_text = [] acked = False deadline = asyncio.get_event_loop().time() + timeout while asyncio.get_event_loop().time() < deadline: remaining = deadline - asyncio.get_event_loop().time() if remaining <= 0: break try: data = await recv_json(ws, timeout=min(remaining, 30)) except asyncio.TimeoutError: if collected_text: logger.info("Timeout with collected text, returning") break continue msg_type = data.get("type") event_name = data.get("event") if msg_type == "res" and data.get("id") == req_id: if data.get("ok"): acked = True logger.info("chat.send acked, runId=%s", data.get("payload", {}).get("runId")) continue else: return f"Error: {json.dumps(data.get('error', {}), ensure_ascii=False)}" if msg_type == "event" and event_name == "session.message": payload = data.get("payload", {}) msg = payload.get("message", {}) role = msg.get("role", "") text = extract_text(msg) logger.info("session.message role=%s text_len=%d", role, len(text)) if role == "assistant" and text: collected_text.append(text) if msg_type == "event" and event_name == "agent" and data.get("payload", {}).get("stream") == "lifecycle": agent_data = data.get("payload", {}).get("data", {}) phase = agent_data.get("phase", "") if phase in ("end", "complete", "error"): logger.info("Agent phase=%s, breaking", phase) if collected_text: break try: for _ in range(5): extra = await recv_json(ws, timeout=3) if extra.get("type") == "event" and extra.get("event") == "session.message": emsg = extra.get("payload", {}).get("message", {}) etext = extract_text(emsg) if emsg.get("role") == "assistant" and etext: collected_text.append(etext) break except asyncio.TimeoutError: pass break if msg_type == "event" and event_name == "sessions.changed": phase = data.get("payload", {}).get("phase", "") if phase == "idle" and collected_text: logger.info("Session idle, returning") break if collected_text: return collected_text[-1] if acked: return "OpenClaw 正在处理中，请稍后查看结果。" return "OpenClaw 连接失败，请重试。" # ==================== Async OpenClaw ==================== HA_TTS_DEVICE = os.getenv("HA_TTS_DEVICE", "media_player.ke_ting_google") _openclaw_cache = { "status": "idle", "command": "", "result": "", } HA_TTS_ENGINE = os.getenv("HA_TTS_ENGINE", "tts.google_ai_tts") def _tts_notify(message): """Send TTS notification to the configured speaker.""" try: data = { "entity_id": HA_TTS_ENGINE, "media_player_entity_id": HA_TTS_DEVICE, "message": message, } ha_request("POST", "services/tts/speak", data) _log(f"TTS notify sent: {message}") except Exception as e: _log(f"TTS notify failed: {e}") async def _openclaw_background(user_command): """Background task: wait for OpenClaw result, then notify via TTS.""" _log(f"Background task started for: {user_command}") try: result = await send_chat_and_wait(user_command, timeout=300) _openclaw_cache["result"] = result _openclaw_cache["status"] = "done" _log(f"Background task done, result length: {len(result)}") _tts_notify("OpenClaw 回复好了，你可以问我结果。") except Exception as e: _openclaw_cache["result"] = f"出错了: {e}" _openclaw_cache["status"] = "error" _log(f"Background task error: {e}") _tts_notify("OpenClaw 处理出错了。") @mcp.tool() async def ask_openclaw(user_command: str) -> str: """ 向 OpenClaw AI 助手发送消息。这是异步操作，会立刻返回，不会等待结果。 仅用于：网络搜索、信息查询、写作、代码、对话等需要AI推理的任务。 不要用这个工具来控制智能家居设备（灯、开关、窗帘、空调等），设备控制请使用其他专用工具。 当 OpenClaw 处理完成后，会通过客厅音箱通知用户。 用户可以随时调用 get_openclaw_result 查看结果。 """ _log(f"TOOL CALLED: ask_openclaw({user_command})") if _openclaw_cache["status"] == "pending": return f"OpenClaw 正在处理上一个请求「{_openclaw_cache['command']}」，请等处理完再发新请求。你可以调用 get_openclaw_result 查看状态。" _openclaw_cache["status"] = "pending" _openclaw_cache["command"] = user_command _openclaw_cache["result"] = "" asyncio.get_event_loop().create_task(_openclaw_background(user_command)) return f"已发送给 OpenClaw：「{user_command}」。处理完成后会通过音箱通知你，你也可以随时问我「结果呢」来查看。" @mcp.tool() def get_openclaw_result() -> str: """ 查看 OpenClaw 的处理结果。当用户问"结果呢"、"OpenClaw回复了吗"、"查看结果"时使用此工具。 """ _log(f"TOOL CALLED: get_openclaw_result, status={_openclaw_cache['status']}") status = _openclaw_cache["status"] command = _openclaw_cache["command"] if status == "idle": return "当前没有待处理的 OpenClaw 请求。" elif status == "pending": return f"OpenClaw 正在处理「{command}」，请稍后再问。" elif status in ("done", "error"): result = _openclaw_cache["result"] _openclaw_cache["status"] = "idle" _openclaw_cache["command"] = "" _openclaw_cache["result"] = "" return f"OpenClaw 对「{command}」的回复：\n\n{result}" return "未知状态" @mcp.tool() def read_openclaw_result(use_tts: bool = True) -> str: """ 让音箱朗读 OpenClaw 的回复。当用户说"念给我听"、"播报结果"时使用此工具。 参数: - use_tts: 是否用音箱播报，默认 True """ _log(f"TOOL CALLED: read_openclaw_result(use_tts={use_tts})") status = _openclaw_cache["status"] if status == "pending": return "OpenClaw 还在处理中，请稍后。" if status == "idle": return "当前没有 OpenClaw 的回复可以播报。" result = _openclaw_cache["result"] if use_tts and result: tts_text = result[:500] _tts_notify(tts_text) _openclaw_cache["status"] = "idle" _openclaw_cache["command"] = "" _openclaw_cache["result"] = "" return "正在通过音箱播报 OpenClaw 的回复。" return result # ==================== Spotify ==================== SPOTIFY_CLIENT_ID = os.getenv("SPOTIFY_CLIENT_ID", "").strip() SPOTIFY_CLIENT_SECRET = os.getenv("SPOTIFY_CLIENT_SECRET", "").strip() SPOTIFY_REFRESH_TOKEN = os.getenv("SPOTIFY_REFRESH_TOKEN", "").strip() _spotify_token_cache = {"token": "", "expires": 0} def _spotify_get_token() -> str: import time now = time.time() if _spotify_token_cache["token"] and now < _spotify_token_cache["expires"] - 60: return _spotify_token_cache["token"] if not SPOTIFY_CLIENT_ID or not SPOTIFY_CLIENT_SECRET or not SPOTIFY_REFRESH_TOKEN: raise RuntimeError("Spotify 未配置，请先在 .env 设置 SPOTIFY_CLIENT_ID / SPOTIFY_CLIENT_SECRET / SPOTIFY_REFRESH_TOKEN") import base64 credentials = base64.b64encode(f"{SPOTIFY_CLIENT_ID}:{SPOTIFY_CLIENT_SECRET}".encode()).decode() data = urllib.parse.urlencode({ "grant_type": "refresh_token", "refresh_token": SPOTIFY_REFRESH_TOKEN, }).encode() req = urllib.request.Request( "https://accounts.spotify.com/api/token", data=data, headers={ "Authorization": f"Basic {credentials}", "Content-Type": "application/x-www-form-urlencoded", }, method="POST", ) with urllib.request.urlopen(req, timeout=10) as resp: result = json.loads(resp.read().decode()) _spotify_token_cache["token"] = result["access_token"] _spotify_token_cache["expires"] = now + result.get("expires_in", 3600) return result["access_token"] def _spotify_api(method: str, path: str, body=None, params: dict = None) -> dict: try: token = _spotify_get_token() except Exception as e: return {"error": str(e)} url = f"https://api.spotify.com/v1{path}" if params: url += "?" + urllib.parse.urlencode(params) _log(f"Spotify API: {method} {url[:100]}") headers = {"Authorization": f"Bearer {token}"} data = None if body is not None: data = json.dumps(body).encode() headers["Content-Type"] = "application/json" req = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=15) as resp: raw = resp.read() return json.loads(raw) if raw else {} except urllib.error.HTTPError as e: err_body = e.read().decode()[:400] _log(f"Spotify API error: {e.code} {err_body}") return {"error": f"HTTP {e.code}: {err_body}"} except Exception as e: _log(f"Spotify API exception: {e}") return {"error": str(e)} def _spotify_check() -> str: if not SPOTIFY_CLIENT_ID or not SPOTIFY_CLIENT_SECRET or not SPOTIFY_REFRESH_TOKEN: return "错误: Spotify 未配置，请先运行 spotify_auth.py 获取凭据并填入 .env 文件" return "" @mcp.tool() def spotify_current() -> str: """ 查看 Spotify 当前正在播放的歌曲。 什么时候调用: - 用户问"现在放的什么歌"、"Spotify 在播什么"、"当前歌曲" """ _log("TOOL CALLED: spotify_current") err = _spotify_check() if err: return err data = _spotify_api("GET", "/me/player/currently-playing") if "error" in data: return f"获取失败: {data['error']}" if not data or data.get("currently_playing_type") == "unknown": return "Spotify 当前没有播放任何内容" item = data.get("item", {}) if not item: return "Spotify 当前没有播放任何内容" name = item.get("name", "未知") artists = ", ".join(a["name"] for a in item.get("artists", [])) album = item.get("album", {}).get("name", "") is_playing = data.get("is_playing", False) progress_ms = data.get("progress_ms", 0) duration_ms = item.get("duration_ms", 0) progress = f"{progress_ms // 60000}:{(progress_ms % 60000) // 1000:02d}" duration = f"{duration_ms // 60000}:{(duration_ms % 60000) // 1000:02d}" status = "▶ 播放中" if is_playing else "⏸ 已暂停" return f"{status}\n🎵 {name}\n👤 {artists}\n💿 {album}\n⏱ {progress} / {duration}" @mcp.tool() def spotify_devices() -> str: """ 列出所有可用的 Spotify 播放设备。 什么时候调用: - 用户问"有哪些 Spotify 设备"、"切换到哪个设备" - 在调用 spotify_play 前不确定设备名时 """ _log("TOOL CALLED: spotify_devices") err = _spotify_check() if err: return err data = _spotify_api("GET", "/me/player/devices") if "error" in data: return f"获取设备失败: {data['error']}" devices = data.get("devices", []) if not devices: return "没有找到可用的 Spotify 设备，请确保 Spotify 客户端已打开" lines = [] for d in devices: active = " ← 当前活跃" if d.get("is_active") else "" vol = d.get("volume_percent", "?") lines.append(f"📱 {d['name']} [{d['type']}] 音量:{vol}%{active}\n ID: {d['id']}") return "\n\n".join(lines) @mcp.tool() def spotify_search(query: str, search_type: str = "track", limit: int = 5) -> str: """ 搜索 Spotify 上的歌曲、歌手、专辑或歌单。 什么时候调用: - 用户说"搜一下"、"找一首歌"、"搜索 Spotify" - 在播放前先搜索确认内容 参数: - query: 搜索词，如 "周杰伦 晴天"、"Taylor Swift"、"lofi chill" - search_type: 搜索类型，可选 track（歌曲）、artist（歌手）、album（专辑）、playlist（歌单），默认 track - limit: 返回结果数量，默认5，最多10 """ _log(f"TOOL CALLED: spotify_search({query}, {search_type}, {limit})") err = _spotify_check() if err: return err valid_types = {"track", "artist", "album", "playlist"} if search_type not in valid_types: search_type = "track" limit = max(1, min(limit, 10)) data = _spotify_api("GET", "/search", params={"q": query, "type": search_type, "limit": limit}) if "error" in data: return f"搜索失败: {data['error']}" type_key = search_type + "s" items = data.get(type_key, {}).get("items", []) if not items: return f"未找到与「{query}」相关的{search_type}" lines = [] for i, item in enumerate(items, 1): uri = item.get("uri", "") if search_type == "track": artists = ", ".join(a["name"] for a in item.get("artists", [])) album = item.get("album", {}).get("name", "") lines.append(f"{i}. 🎵 {item['name']}\n 👤 {artists} 💿 {album}\n URI: {uri}") elif search_type == "artist": followers = item.get("followers", {}).get("total", 0) lines.append(f"{i}. 👤 {item['name']} 粉丝: {followers:,}\n URI: {uri}") elif search_type == "album": artists = ", ".join(a["name"] for a in item.get("artists", [])) year = item.get("release_date", "")[:4] lines.append(f"{i}. 💿 {item['name']} ({year})\n 👤 {artists}\n URI: {uri}") elif search_type == "playlist": owner = item.get("owner", {}).get("display_name", "") tracks = item.get("tracks", {}).get("total", "?") lines.append(f"{i}. 📋 {item['name']} by {owner} 共{tracks}首\n URI: {uri}") return f"搜索「{query}」({search_type}) 结果:\n\n" + "\n\n".join(lines) @mcp.tool() def spotify_play(query: str = "", uri: str = "", device_name: str = "") -> str: """ 在 Spotify 上播放歌曲、歌单、专辑或歌手。 什么时候调用: - 用户说"播放 xxx"、"放一首 xxx"、"Spotify 播 xxx" 参数: - query: 搜索关键词，如 "周杰伦 晴天"、"Taylor Swift folklore"。填了 uri 则忽略此项。 - uri: Spotify URI，格式如 spotify:track:xxx 或 spotify:playlist:xxx。优先使用 URI。 - device_name: 指定设备名称（模糊匹配），留空则在当前活跃设备播放。可先用 spotify_devices 查看设备。 """ _log(f"TOOL CALLED: spotify_play(query={query}, uri={uri}, device={device_name})") err = _spotify_check() if err: return err # 找设备 ID device_id = None if device_name: dev_data = _spotify_api("GET", "/me/player/devices") if "error" not in dev_data: for d in dev_data.get("devices", []): if device_name.lower() in d["name"].lower(): device_id = d["id"] break if not device_id: return f"未找到名称包含「{device_name}」的设备，可用 spotify_devices 查看设备列表" # 如果没有 URI，先搜索 if not uri and query: search_data = _spotify_api("GET", "/search", params={"q": query, "type": "track", "limit": 1}) if "error" in search_data: return f"搜索失败: {search_data['error']}" items = search_data.get("tracks", {}).get("items", []) if not items: # 尝试搜索歌单 search_data2 = _spotify_api("GET", "/search", params={"q": query, "type": "playlist", "limit": 1}) items2 = search_data2.get("playlists", {}).get("items", []) if items2: uri = items2[0]["uri"] found_name = items2[0]["name"] else: return f"未找到与「{query}」相关的内容" else: uri = items[0]["uri"] found_name = items[0]["name"] + " - " + ", ".join(a["name"] for a in items[0].get("artists", [])) else: found_name = uri if not uri: return "请提供搜索词或 Spotify URI" # 构建播放请求 body = {} if uri.startswith("spotify:track:"): body["uris"] = [uri] else: body["context_uri"] = uri path = "/me/player/play" params = {} if device_id: params["device_id"] = device_id result = _spotify_api("PUT", path + ("?" + urllib.parse.urlencode(params) if params else ""), body=body) if "error" in result: err_msg = result["error"] if "NO_ACTIVE_DEVICE" in str(err_msg) or "404" in str(err_msg): return f"没有活跃的 Spotify 设备，请先打开 Spotify 客户端，或用 spotify_devices 查看设备后用 device_name 参数指定" return f"播放失败: {err_msg}" return f"▶ 正在播放: {found_name}" @mcp.tool() def spotify_control(action: str) -> str: """ 控制 Spotify 播放状态。 什么时候调用: - 用户说"暂停"、"继续播放"、"下一首"、"上一首"、"停止" 参数: - action: 操作类型: - pause: 暂停 - resume / play: 继续播放 - next: 下一首 - previous: 上一首 - shuffle_on: 开启随机播放 - shuffle_off: 关闭随机播放 - repeat_track: 单曲循环 - repeat_context: 列表循环 - repeat_off: 关闭循环 """ _log(f"TOOL CALLED: spotify_control({action})") err = _spotify_check() if err: return err action_map = { "pause": ("PUT", "/me/player/pause", None, None), "resume": ("PUT", "/me/player/play", None, None), "play": ("PUT", "/me/player/play", None, None), "next": ("POST", "/me/player/next", None, None), "previous": ("POST", "/me/player/previous", None, None), "shuffle_on": ("PUT", "/me/player/shuffle", None, {"state": "true"}), "shuffle_off": ("PUT", "/me/player/shuffle", None, {"state": "false"}), "repeat_track": ("PUT", "/me/player/repeat", None, {"state": "track"}), "repeat_context": ("PUT", "/me/player/repeat", None, {"state": "context"}), "repeat_off": ("PUT", "/me/player/repeat", None, {"state": "off"}), } if action not in action_map: return f"不支持的操作: {action}。可选: pause/resume/play/next/previous/shuffle_on/shuffle_off/repeat_track/repeat_context/repeat_off" method, path, body, params = action_map[action] if params: path = path + "?" + urllib.parse.urlencode(params) result = _spotify_api(method, path, body=body) if "error" in result: return f"操作失败: {result['error']}" action_names = { "pause": "已暂停", "resume": "已继续播放", "play": "已继续播放", "next": "已切到下一首", "previous": "已切到上一首", "shuffle_on": "随机播放已开启", "shuffle_off": "随机播放已关闭", "repeat_track": "单曲循环已开启", "repeat_context": "列表循环已开启", "repeat_off": "循环已关闭", } return action_names.get(action, f"已执行: {action}") @mcp.tool() def spotify_volume(volume: int) -> str: """ 设置 Spotify 播放音量。 参数: - volume: 音量百分比，0-100 """ _log(f"TOOL CALLED: spotify_volume({volume})") err = _spotify_check() if err: return err volume = max(0, min(100, volume)) result = _spotify_api("PUT", f"/me/player/volume?volume_percent={volume}") if "error" in result: return f"设置音量失败: {result['error']}" return f"Spotify 音量已设为 {volume}%" @mcp.tool() def spotify_queue_add(query: str = "", uri: str = "") -> str: """ 把一首歌加入 Spotify 播放队列。 参数: - query: 搜索歌曲名，如 "周杰伦 青花瓷" - uri: Spotify track URI（优先使用），如 spotify:track:xxx """ _log(f"TOOL CALLED: spotify_queue_add(query={query}, uri={uri})") err = _spotify_check() if err: return err if not uri and query: search_data = _spotify_api("GET", "/search", params={"q": query, "type": "track", "limit": 1}) if "error" in search_data: return f"搜索失败: {search_data['error']}" items = search_data.get("tracks", {}).get("items", []) if not items: return f"未找到「{query}」" uri = items[0]["uri"] found_name = items[0]["name"] + " - " + ", ".join(a["name"] for a in items[0].get("artists", [])) else: found_name = uri if not uri: return "请提供搜索词或 Spotify URI" result = _spotify_api("POST", f"/me/player/queue?uri={urllib.parse.quote(uri)}") if "error" in result: return f"加入队列失败: {result['error']}" return f"已加入播放队列: {found_name}" # ==================== ERP ==================== ERP_URL = os.getenv("ERP_URL", "http://localhost:8000").rstrip("/") ERP_USERNAME = os.getenv("ERP_USERNAME", "") ERP_PASSWORD = os.getenv("ERP_PASSWORD", "") _erp_jwt_token: str = "" _erp_sessions: dict[str, list[dict]] = {} _ERP_SESSION_CONTEXT_LIMIT = 16 _ERP_REPLY_LIMIT = 900 def _erp_http(method: str, path: str, body: Any = None, token: str = "") -> dict: url = f"{ERP_URL}{path}" data = json.dumps(body).encode() if body is not None else None headers: dict[str, str] = {"Content-Type": "application/json"} if token: headers["Authorization"] = f"Bearer {token}" req = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=20) as resp: return json.loads(resp.read().decode()) except urllib.error.HTTPError as e: body_text = e.read().decode() if e.fp else "" return {"error": f"HTTP {e.code}", "detail": body_text[:200], "status_code": e.code} except Exception as e: return {"error": str(e)} def _erp_login() -> str: if not ERP_USERNAME or not ERP_PASSWORD: logger.error("ERP_USERNAME / ERP_PASSWORD not set") return "" result = _erp_http("POST", "/api/auth/login", {"username": ERP_USERNAME, "password": ERP_PASSWORD}) token = result.get("access_token", "") if token: logger.info("ERP login successful") else: logger.error("ERP login failed: %s", result.get("error") or result) return token def _erp_ensure_token() -> str: global _erp_jwt_token if not _erp_jwt_token: _erp_jwt_token = _erp_login() return _erp_jwt_token def _erp_ai_chat(message: str, context: list[dict]) -> str: global _erp_jwt_token payload = {"message": message, "context": context} for _attempt in range(2): token = _erp_ensure_token() if not token: return "ERP 登录失败，请检查 ERP_USERNAME / ERP_PASSWORD 配置。" result = _erp_http("POST", "/api/ai/chat", payload, token=token) if result.get("error") and result.get("status_code") == 401: logger.warning("ERP JWT expired, re-logging in") _erp_jwt_token = "" continue if result.get("error"): logger.error("ERP ai/chat error: %s", result) return f"ERP 接口错误: {result.get('error')}" reply = result.get("reply") or "（无回复）" logger.info("ERP ai/chat ok, reply_len=%d", len(reply)) return reply return "ERP 登录失败，无法获取有效 token。" def _erp_get_context(session_id: str) -> list[dict]: return _erp_sessions.get(session_id, []) def _erp_save_context(session_id: str, user_msg: str, assistant_reply: str) -> None: ctx = _erp_sessions.setdefault(session_id, []) ctx.extend([ {"role": "user", "content": user_msg}, {"role": "assistant", "content": assistant_reply}, ]) if len(ctx) > _ERP_SESSION_CONTEXT_LIMIT: _erp_sessions[session_id] = ctx[-_ERP_SESSION_CONTEXT_LIMIT:] def _erp_truncate(text: str) -> str: encoded = text.encode() if len(encoded) <= _ERP_REPLY_LIMIT: return text return encoded[:_ERP_REPLY_LIMIT].decode(errors="ignore") + "..." @mcp.tool() def chat_with_erp(message: str, session_id: str = "default") -> str: """ 向 ERP 系统发送消息，查询业务数据或执行操作。 当用户询问发票、客户、收入、支出、订阅、GST 税务、票据等 ERP 相关信息时使用此工具。 也可以通过自然语言创建发票草稿或新建客户资料。 参数: - message: 用户的完整问题或指令，例如"查询本月未付款发票"、"上个月收入多少"。 - session_id: 会话标识，用于保持多轮对话上下文，同一次对话请传相同的值，默认 default。 """ _log(f"TOOL CALLED: chat_with_erp({message}, session_id={session_id})") ctx = _erp_get_context(session_id) reply = _erp_ai_chat(message, ctx) _erp_save_context(session_id, message, reply) return _erp_truncate(reply) @mcp.tool() def reset_erp_session(session_id: str = "default") -> str: """ 清除指定 ERP 会话的对话历史，开启新一轮对话。 当用户说"重新开始"、"清除记录"、"新对话"且上下文是 ERP 查询时使用此工具。 """ _erp_sessions.pop(session_id, None) logger.info("ERP session %s reset", session_id) return "已清除 ERP 对话记录，可以开始新的 ERP 对话。" if __name__ == "__main__": import sys mode = sys.argv[1] if len(sys.argv) > 1 else "stdio" if mode == "sse": logger.info("Starting MCP SSE server on %s:%d", BRIDGE_HOST, BRIDGE_PORT) mcp.run(transport="sse") else: mcp.run(transport="stdio")