add tool research in examples

agent/core/runner.py

@@ -165,6 +165,10 @@ BUILTIN_TOOLS = [
     "search_posts",
     "select_post",
     "get_search_suggestions",
+    "x_search",
+    "youtube_search",
+    "youtube_detail",
+    "import_content",
 
     # 知识管理工具
     "knowledge_search",
@@ -216,6 +220,14 @@ BUILTIN_TOOLS = [
     "feishu_get_chat_history",
     "feishu_get_contact_replies",
     "feishu_get_contact_list",
+
+    # IM 工具
+    "im_setup",
+    "im_check_notification",
+    "im_receive_messages",
+    "im_send_message",
+    "im_get_contacts",
+    "im_get_chat_history",
 ]
 
 
@@ -2354,7 +2366,7 @@ class AgentRunner:
                     new_content.append(block)
 
             msg["content"] = new_content
-        print(f"[Image Opt Check] 扫描到 {stats['kept'] + stats['downscaled'] + stats['described']} 张图片上下文")
+        # print(f"[Image Opt Check] 扫描到 {stats['kept'] + stats['downscaled'] + stats['described']} 张图片上下文")
         if stats["downscaled"] > 0 or stats["described"] > 0:
             logger.info(
                 f"[Image Optimization] 保留 {stats['kept']} 张，"

agent/tools/builtin/__init__.py

@@ -21,11 +21,14 @@ from agent.tools.builtin.sandbox import (sandbox_create_environment, sandbox_run
 from agent.tools.builtin.knowledge import(knowledge_search,knowledge_save,knowledge_list,knowledge_update,knowledge_batch_update,knowledge_slim)
 from agent.tools.builtin.context import get_current_context
 from agent.tools.builtin.toolhub import toolhub_health, toolhub_search, toolhub_call, toolhub_create
+from agent.tools.builtin.resource import resource_list_tools, resource_get_tool
+from agent.tools.builtin.crawler import youtube_search, youtube_detail, x_search, import_content, extract_video_clip
 from agent.trace.goal_tool import goal
 # 导入浏览器工具以触发注册
 import agent.tools.builtin.browser  # noqa: F401
 
 import agent.tools.builtin.feishu
+import agent.tools.builtin.im
 
 __all__ = [
     # 文件操作
@@ -60,6 +63,15 @@ __all__ = [
     "toolhub_search",
     "toolhub_call",
     "toolhub_create",
+    # 资源查询
+    "resource_list_tools",
+    "resource_get_tool",
+    # 爬虫工具
+    "youtube_search",
+    "youtube_detail",
+    "x_search",
+    "import_content",
+    "extract_video_clip",
     # Goal 管理
     "goal",
 ]

agent/tools/builtin/crawler.py

@@ -0,0 +1,595 @@
+"""
+爬虫服务工具模块
+
+提供 YouTube、X (Twitter) 和微信/通用链接的搜索和详情查询功能。
+"""
+
+import asyncio
+import base64
+import io
+import json
+import math
+import os
+import subprocess
+import tempfile
+from pathlib import Path
+from typing import Optional, List, Dict, Any
+
+import httpx
+from PIL import Image, ImageDraw, ImageFont
+
+from agent.tools import tool, ToolResult
+
+
+# API 配置
+CRAWLER_BASE_URL = "http://crawler.aiddit.com/crawler"
+AIGC_BASE_URL = "http://aigc-channel.aiddit.com/aigc/channel"
+DEFAULT_TIMEOUT = 60.0
+
+# 拼接图配置
+THUMB_WIDTH = 250
+THUMB_HEIGHT = 250
+TEXT_HEIGHT = 80
+GRID_COLS = 5
+PADDING = 12
+BG_COLOR = (255, 255, 255)
+TEXT_COLOR = (30, 30, 30)
+INDEX_COLOR = (220, 60, 60)
+
+# 视频处理相关配置
+VIDEO_DOWNLOAD_DIR = Path(tempfile.gettempdir()) / "youtube_videos"
+VIDEO_DOWNLOAD_DIR.mkdir(exist_ok=True)
+
+
+# ── 辅助函数 ──
+
+def _truncate_text(text: str, max_len: int = 14) -> str:
+    """截断文本，超出部分用省略号"""
+    return text[:max_len] + "..." if len(text) > max_len else text
+
+
+async def _download_image(client: httpx.AsyncClient, url: str) -> Optional[Image.Image]:
+    """下载单张图片，失败返回 None"""
+    try:
+        resp = await client.get(url, timeout=15.0)
+        resp.raise_for_status()
+        return Image.open(io.BytesIO(resp.content)).convert("RGB")
+    except Exception:
+        return None
+
+
+async def _build_video_collage(videos: List[Dict[str, Any]]) -> Optional[str]:
+    """
+    将视频缩略图+序号+标题拼接成网格图，返回 base64 编码的 PNG。
+    """
+    if not videos:
+        return None
+
+    items = []
+    for idx, video in enumerate(videos):
+        thumbnail = None
+        if "thumbnails" in video and isinstance(video["thumbnails"], list) and video["thumbnails"]:
+            thumbnail = video["thumbnails"][0].get("url")
+        elif "thumbnail" in video:
+            thumbnail = video.get("thumbnail")
+        elif "cover_url" in video:
+            thumbnail = video.get("cover_url")
+
+        title = video.get("title", "") or video.get("text", "")
+        if thumbnail:
+            items.append({"url": thumbnail, "title": title, "index": idx + 1})
+    if not items:
+        return None
+
+    async with httpx.AsyncClient() as client:
+        tasks = [_download_image(client, item["url"]) for item in items]
+        downloaded = await asyncio.gather(*tasks)
+
+    valid = [(item, img) for item, img in zip(items, downloaded) if img is not None]
+    if not valid:
+        return None
+
+    cols = min(GRID_COLS, len(valid))
+    rows = math.ceil(len(valid) / cols)
+    cell_w = THUMB_WIDTH + PADDING
+    cell_h = THUMB_HEIGHT + TEXT_HEIGHT + PADDING
+    canvas_w = cols * cell_w + PADDING
+    canvas_h = rows * cell_h + PADDING
+
+    canvas = Image.new("RGB", (canvas_w, canvas_h), BG_COLOR)
+    draw = ImageDraw.Draw(canvas)
+
+    font_title = None
+    font_index = None
+    font_candidates = [
+        "msyh.ttc", "simhei.ttf", "simsun.ttc",
+        "/System/Library/Fonts/PingFang.ttc",
+        "/usr/share/fonts/truetype/droid/DroidSansFallbackFull.ttf",
+        "/usr/share/fonts/truetype/wqy/wqy-microhei.ttc",
+        "/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc",
+    ]
+    for font_path in font_candidates:
+        try:
+            font_title = ImageFont.truetype(font_path, 16)
+            font_index = ImageFont.truetype(font_path, 32)
+            break
+        except Exception:
+            continue
+    if not font_title:
+        font_title = ImageFont.load_default()
+        font_index = font_title
+
+    for item, img in valid:
+        idx = item["index"]
+        col = (idx - 1) % cols
+        row = (idx - 1) // cols
+        x = PADDING + col * cell_w
+        y = PADDING + row * cell_h
+
+        scale = min(THUMB_WIDTH / img.width, THUMB_HEIGHT / img.height)
+        new_w = int(img.width * scale)
+        new_h = int(img.height * scale)
+        thumb = img.resize((new_w, new_h), Image.LANCZOS)
+        offset_x = x + (THUMB_WIDTH - new_w) // 2
+        offset_y = y + (THUMB_HEIGHT - new_h) // 2
+        canvas.paste(thumb, (offset_x, offset_y))
+
+        index_text = str(idx)
+        idx_x = offset_x
+        idx_y = offset_y + 4
+        box_size = 52
+        draw.rectangle([idx_x, idx_y, idx_x + box_size, idx_y + box_size], fill=INDEX_COLOR)
+        bbox = draw.textbbox((0, 0), index_text, font=font_index)
+        tw, th = bbox[2] - bbox[0], bbox[3] - bbox[1]
+        text_x = idx_x + (box_size - tw) // 2
+        text_y = idx_y + (box_size - th) // 2
+        draw.text((text_x, text_y), index_text, fill=(255, 255, 255), font=font_index)
+
+        title = item["title"] or ""
+        if title:
+            words = list(title)
+            lines = []
+            current_line = ""
+            for ch in words:
+                test_line = current_line + ch
+                bbox_line = draw.textbbox((0, 0), test_line, font=font_title)
+                if bbox_line[2] - bbox_line[0] > THUMB_WIDTH:
+                    if current_line:
+                        lines.append(current_line)
+                    current_line = ch
+                else:
+                    current_line = test_line
+            if current_line:
+                lines.append(current_line)
+            for line_i, line in enumerate(lines):
+                draw.text((x, y + THUMB_HEIGHT + 6 + line_i * 22), line, fill=TEXT_COLOR, font=font_title)
+
+    buf = io.BytesIO()
+    canvas.save(buf, format="PNG")
+    return base64.b64encode(buf.getvalue()).decode("utf-8")
+
+
+def _parse_srt_to_outline(srt_content: str) -> List[Dict[str, str]]:
+    """解析 SRT 字幕，生成带时间戳的大纲"""
+    if not srt_content:
+        return []
+
+    outline = []
+    blocks = srt_content.strip().split('\n\n')
+    for block in blocks:
+        lines = block.strip().split('\n')
+        if len(lines) >= 3:
+            timestamp_line = lines[1]
+            if '-->' in timestamp_line:
+                start_time = timestamp_line.split('-->')[0].strip()
+                text = ' '.join(lines[2:])
+                outline.append({'timestamp': start_time, 'text': text})
+    return outline
+
+
+def _download_youtube_video(video_id: str) -> Optional[str]:
+    """使用 yt-dlp 下载 YouTube 视频，返回文件路径"""
+    try:
+        output_path = VIDEO_DOWNLOAD_DIR / f"{video_id}.mp4"
+        if output_path.exists():
+            return str(output_path)
+
+        cmd = [
+            'yt-dlp',
+            '-f', 'best[ext=mp4]',
+            '-o', str(output_path),
+            f'https://www.youtube.com/watch?v={video_id}'
+        ]
+        result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
+
+        if result.returncode == 0 and output_path.exists():
+            return str(output_path)
+        return None
+    except Exception:
+        return None
+
+
+# ── YouTube 工具 ──
+
+@tool()
+async def youtube_search(keyword: str) -> ToolResult:
+    """
+    搜索 YouTube 视频
+
+    Args:
+        keyword: 搜索关键词
+
+    Returns:
+        搜索结果列表，包含视频标题、ID、频道等信息
+    """
+    try:
+        async with httpx.AsyncClient(timeout=DEFAULT_TIMEOUT) as client:
+            response = await client.post(
+                f"{CRAWLER_BASE_URL}/youtube/keyword",
+                json={"keyword": keyword}
+            )
+            response.raise_for_status()
+            data = response.json()
+
+            if data.get("code") == 0:
+                result_data = data.get("data", {})
+                videos = result_data.get("data", []) if isinstance(result_data, dict) else []
+
+                images = []
+                collage_b64 = await _build_video_collage(videos)
+                if collage_b64:
+                    images.append({
+                        "type": "base64",
+                        "media_type": "image/png",
+                        "data": collage_b64
+                    })
+
+                summary_list = []
+                for idx, video in enumerate(videos[:20], 1):
+                    title = video.get("title", "")
+                    author = video.get("author", "")
+                    video_id = video.get("video_id", "")
+                    summary_list.append(f"{idx}. {title} - {author} (ID: {video_id})")
+
+                output_data = {
+                    "keyword": keyword,
+                    "total": len(videos),
+                    "summary": summary_list,
+                    "data": videos
+                }
+
+                return ToolResult(
+                    title=f"YouTube 搜索: {keyword}",
+                    output=json.dumps(output_data, ensure_ascii=False, indent=2),
+                    long_term_memory=f"Searched YouTube for '{keyword}', found {len(videos)} videos",
+                    images=images
+                )
+            else:
+                return ToolResult(
+                    title="YouTube 搜索失败",
+                    output="",
+                    error=f"搜索失败: {data.get('msg', '未知错误')}"
+                )
+
+    except Exception as e:
+        return ToolResult(
+            title="YouTube 搜索异常",
+            output="",
+            error=str(e)
+        )
+
+
+@tool()
+async def youtube_detail(
+    content_id: str,
+    include_captions: bool = True,
+    download_video: bool = False
+) -> ToolResult:
+    """
+    获取 YouTube 视频详情（可选包含字幕、下载视频并生成大纲）
+
+    Args:
+        content_id: 视频 ID
+        include_captions: 是否包含字幕，默认 True
+        download_video: 是否下载视频并生成带时间戳的大纲，默认 False。
+            下载后可使用 extract_video_clip 截取视频片段观看。
+
+    Returns:
+        视频详细信息，包含字幕、视频大纲和本地文件路径
+    """
+    try:
+        async with httpx.AsyncClient(timeout=DEFAULT_TIMEOUT) as client:
+            detail_response = await client.post(
+                f"{CRAWLER_BASE_URL}/youtube/detail",
+                json={"content_id": content_id}
+            )
+            detail_response.raise_for_status()
+            detail_data = detail_response.json()
+
+            if detail_data.get("code") != 0:
+                return ToolResult(
+                    title="获取详情失败",
+                    output="",
+                    error=f"获取详情失败: {detail_data.get('msg', '未知错误')}"
+                )
+
+            result_data = detail_data.get("data", {})
+            video_info = result_data.get("data", {}) if isinstance(result_data, dict) else {}
+
+            # 获取字幕
+            captions_text = None
+            if include_captions or download_video:
+                try:
+                    captions_response = await client.post(
+                        f"{CRAWLER_BASE_URL}/youtube/captions",
+                        json={"content_id": content_id}
+                    )
+                    captions_response.raise_for_status()
+                    captions_data = captions_response.json()
+
+                    if captions_data.get("code") == 0:
+                        captions_result = captions_data.get("data", {})
+                        if isinstance(captions_result, dict):
+                            inner_data = captions_result.get("data", {})
+                            if isinstance(inner_data, dict):
+                                captions_text = inner_data.get("content")
+                except Exception:
+                    pass
+
+            # 下载视频并生成大纲
+            video_path = None
+            video_outline = None
+            if download_video:
+                video_path = await asyncio.to_thread(_download_youtube_video, content_id)
+                if captions_text:
+                    video_outline = _parse_srt_to_outline(captions_text)
+
+            # 合并数据
+            output_data = {
+                "video_id": content_id,
+                "title": video_info.get("title", ""),
+                "channel": video_info.get("channel_account_name", ""),
+                "description": video_info.get("body_text", ""),
+                "like_count": video_info.get("like_count"),
+                "comment_count": video_info.get("comment_count"),
+                "publish_timestamp": video_info.get("publish_timestamp"),
+                "content_link": video_info.get("content_link", ""),
+                "captions": captions_text,
+                "full_data": video_info
+            }
+
+            if download_video:
+                output_data["video_path"] = video_path
+                output_data["video_outline"] = video_outline
+                if not video_path:
+                    output_data["download_error"] = "视频下载失败，请检查 yt-dlp 是否可用"
+
+            memory = f"Retrieved YouTube video details for {content_id}"
+            if captions_text:
+                memory += " with captions"
+            if video_path:
+                memory += f", downloaded to {video_path}"
+
+            return ToolResult(
+                title=f"YouTube 视频详情: {content_id}",
+                output=json.dumps(output_data, ensure_ascii=False, indent=2),
+                long_term_memory=memory
+            )
+
+    except Exception as e:
+        return ToolResult(
+            title="YouTube 详情查询异常",
+            output="",
+            error=str(e)
+        )
+
+
+# ── X (Twitter) 工具 ──
+
+@tool()
+async def x_search(keyword: str) -> ToolResult:
+    """
+    搜索 X (Twitter) 内容（数据已结构化，无需访问详情页）
+
+    Args:
+        keyword: 搜索关键词
+
+    Returns:
+        搜索结果列表，包含推文内容、作者、互动数据等
+    """
+    try:
+        async with httpx.AsyncClient(timeout=DEFAULT_TIMEOUT) as client:
+            response = await client.post(
+                "http://crawler.aiddit.com/crawler/x/keyword",
+                json={"keyword": keyword}
+            )
+            response.raise_for_status()
+            data = response.json()
+
+            if data.get("code") == 0:
+                result_data = data.get("data", {})
+                tweets = result_data.get("data", []) if isinstance(result_data, dict) else []
+
+                # 构建拼接图
+                images = []
+                tweets_with_images = []
+                for tweet in tweets:
+                    image_list = tweet.get("image_url_list", [])
+                    if image_list:
+                        tweet["thumbnails"] = [{"url": image_list[0].get("image_url")}]
+                        tweets_with_images.append(tweet)
+
+                collage_b64 = await _build_video_collage(tweets_with_images if tweets_with_images else tweets)
+                if collage_b64:
+                    images.append({
+                        "type": "base64",
+                        "media_type": "image/png",
+                        "data": collage_b64
+                    })
+
+                summary_list = []
+                for idx, tweet in enumerate(tweets[:20], 1):
+                    text = tweet.get("body_text", "")[:100]
+                    author = tweet.get("channel_account_name", "")
+                    summary_list.append(f"{idx}. @{author}: {text}")
+
+                output_data = {
+                    "keyword": keyword,
+                    "total": len(tweets),
+                    "summary": summary_list,
+                    "data": tweets
+                }
+
+                return ToolResult(
+                    title=f"X 搜索: {keyword}",
+                    output=json.dumps(output_data, ensure_ascii=False, indent=2),
+                    long_term_memory=f"Searched X (Twitter) for '{keyword}', found {len(tweets)} tweets",
+                    images=images
+                )
+            else:
+                return ToolResult(
+                    title="X 搜索失败",
+                    output="",
+                    error=f"搜索失败: {data.get('msg', '未知错误')}"
+                )
+
+    except Exception as e:
+        return ToolResult(
+            title="X 搜索异常",
+            output="",
+            error=str(e)
+        )
+
+
+# ── 内容导入工具 ──
+
+@tool()
+async def import_content(plan_name: str, content_data: List[Dict[str, Any]]) -> ToolResult:
+    """
+    导入长文内容（微信公众号、小红书、抖音等通用链接）
+
+    Args:
+        plan_name: 计划名称
+        content_data: 内容数据列表，每项包含 channel、content_link、title 等字段
+
+    Returns:
+        导入结果，包含 plan_id
+    """
+    try:
+        async with httpx.AsyncClient(timeout=DEFAULT_TIMEOUT) as client:
+            response = await client.post(
+                f"{AIGC_BASE_URL}/weixin/auto_insert",
+                json={"plan_name": plan_name, "data": content_data}
+            )
+            response.raise_for_status()
+            data = response.json()
+
+            if data.get("code") == 0:
+                result_data = data.get("data", {})
+                return ToolResult(
+                    title=f"内容导入: {plan_name}",
+                    output=json.dumps(result_data, ensure_ascii=False, indent=2),
+                    long_term_memory=f"Imported {len(content_data)} items to plan '{plan_name}'"
+                )
+            else:
+                return ToolResult(
+                    title="导入失败",
+                    output="",
+                    error=f"导入失败: {data.get('msg', '未知错误')}"
+                )
+
+    except Exception as e:
+        return ToolResult(
+            title="内容导入异常",
+            output="",
+            error=str(e)
+        )
+
+
+# ── 视频截取工具 ──
+
+@tool()
+async def extract_video_clip(
+    video_id: str,
+    start_time: str,
+    end_time: str,
+    output_name: Optional[str] = None
+) -> ToolResult:
+    """
+    从已下载的 YouTube 视频中截取指定时间段的片段
+
+    Args:
+        video_id: YouTube 视频 ID（必须先通过 youtube_detail(download_video=True) 下载）
+        start_time: 开始时间，格式: HH:MM:SS 或 MM:SS
+        end_time: 结束时间，格式: HH:MM:SS 或 MM:SS
+        output_name: 输出文件名（可选）
+
+    Returns:
+        截取的视频片段路径
+
+    Example:
+        extract_video_clip("dQw4w9WgXcQ", "00:00:10", "00:00:30")
+    """
+    try:
+        source_video = VIDEO_DOWNLOAD_DIR / f"{video_id}.mp4"
+        if not source_video.exists():
+            return ToolResult(
+                title="视频截取失败",
+                output="",
+                error="源视频不存在，请先使用 youtube_detail(download_video=True) 下载视频"
+            )
+
+        if not output_name:
+            output_name = f"{video_id}_clip_{start_time.replace(':', '-')}_{end_time.replace(':', '-')}.mp4"
+
+        output_path = VIDEO_DOWNLOAD_DIR / output_name
+
+        cmd = [
+            'ffmpeg',
+            '-i', str(source_video),
+            '-ss', start_time,
+            '-to', end_time,
+            '-c', 'copy',
+            '-y',
+            str(output_path)
+        ]
+
+        result = await asyncio.to_thread(
+            subprocess.run, cmd, capture_output=True, text=True, timeout=60
+        )
+
+        if result.returncode == 0 and output_path.exists():
+            file_size = output_path.stat().st_size / (1024 * 1024)
+
+            output_data = {
+                "video_id": video_id,
+                "clip_path": str(output_path),
+                "start_time": start_time,
+                "end_time": end_time,
+                "file_size_mb": round(file_size, 2)
+            }
+
+            return ToolResult(
+                title=f"视频片段截取成功: {start_time} - {end_time}",
+                output=json.dumps(output_data, ensure_ascii=False, indent=2),
+                long_term_memory=f"Extracted video clip from {video_id}: {start_time} to {end_time}"
+            )
+        else:
+            return ToolResult(
+                title="视频截取失败",
+                output="",
+                error=f"ffmpeg 执行失败: {result.stderr}"
+            )
+
+    except subprocess.TimeoutExpired:
+        return ToolResult(
+            title="视频截取超时",
+            output="",
+            error="视频截取超时（60秒）"
+        )
+    except Exception as e:
+        return ToolResult(
+            title="视频截取异常",
+            output="",
+            error=str(e)
+        )

agent/tools/builtin/im/__init__.py

@@ -0,0 +1,17 @@
+from agent.tools.builtin.im.chat import (
+    im_setup,
+    im_check_notification,
+    im_receive_messages,
+    im_send_message,
+    im_get_contacts,
+    im_get_chat_history,
+)
+
+__all__ = [
+    "im_setup",
+    "im_check_notification",
+    "im_receive_messages",
+    "im_send_message",
+    "im_get_contacts",
+    "im_get_chat_history",
+]

agent/tools/builtin/im/chat.py

@@ -0,0 +1,329 @@
+"""IM 工具 — 将 im-client 接入 Agent 框架。
+
+新架构：一个 Agent (contact_id) = 一个 IMClient 实例，该实例管理多个窗口 (chat_id)。
+"""
+
+import asyncio
+import json
+import logging
+import os
+import sys
+from typing import Optional
+
+from agent.tools import tool, ToolResult, ToolContext
+
+# 将 im-client 目录加入 sys.path
+_IM_CLIENT_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "..", "..", "im-client"))
+if _IM_CLIENT_DIR not in sys.path:
+    sys.path.insert(0, _IM_CLIENT_DIR)
+
+from client import IMClient  # noqa: E402
+from notifier import AgentNotifier  # noqa: E402
+
+logger = logging.getLogger(__name__)
+
+# ── 全局状态 ──
+
+_clients: dict[str, IMClient] = {}
+_tasks: dict[str, asyncio.Task] = {}
+_notifications: dict[tuple[str, str], dict] = {}  # (contact_id, chat_id) -> 通知
+
+
+class _ToolNotifier(AgentNotifier):
+    """内部通知器：按 (contact_id, chat_id) 分发通知。"""
+
+    def __init__(self, contact_id: str, chat_id: str):
+        self._key = (contact_id, chat_id)
+
+    async def notify(self, count: int, from_contacts: list[str]):
+        _notifications[self._key] = {"count": count, "from": from_contacts}
+
+
+# ── Tool 1: 初始化连接 ──
+
+@tool(
+    hidden_params=["context"],
+    display={
+        "zh": {"name": "初始化 IM 连接", "params": {"contact_id": "你的身份 ID", "server_url": "服务器地址"}},
+        "en": {"name": "Setup IM Connection", "params": {"contact_id": "Your identity ID", "server_url": "Server URL"}},
+    }
+)
+async def im_setup(
+    contact_id: str,
+    server_url: str = "ws://localhost:8000",
+    notify_interval: float = 10.0,
+    context: Optional[ToolContext] = None,
+) -> ToolResult:
+    """初始化 IM Client 并连接 Server。
+
+    Args:
+        contact_id: 你在 IM 系统中的身份 ID
+        server_url: Server 的 WebSocket 地址
+        notify_interval: 检查新消息的间隔秒数
+    """
+    if contact_id in _clients:
+        return ToolResult(title="IM 已连接", output=f"已连接: {contact_id}")
+
+    client = IMClient(contact_id=contact_id, server_url=server_url, notify_interval=notify_interval)
+    _clients[contact_id] = client
+
+    loop = asyncio.get_event_loop()
+    _tasks[contact_id] = loop.create_task(client.run())
+
+    # 等待一小段时间，让连接尝试开始（非阻塞）
+    await asyncio.sleep(0.5)
+
+    return ToolResult(title="IM 连接成功", output=f"已启动 IM Client: {contact_id}，后台连接中...")
+
+
+# ── Tool 2: 窗口管理 ──
+
+@tool(
+    hidden_params=["context"],
+    display={
+        "zh": {"name": "打开 IM 窗口", "params": {"contact_id": "Agent ID", "chat_id": "窗口 ID"}},
+        "en": {"name": "Open IM Window", "params": {"contact_id": "Agent ID", "chat_id": "Window ID"}},
+    }
+)
+async def im_open_window(
+    contact_id: str,
+    chat_id: str | None = None,
+    context: Optional[ToolContext] = None,
+) -> ToolResult:
+    """打开一个新的聊天窗口。
+
+    Args:
+        contact_id: Agent ID
+        chat_id: 窗口 ID（留空自动生成）
+    """
+    client = _clients.get(contact_id)
+    if client is None:
+        return ToolResult(title="未连接", output=f"错误: {contact_id} 未初始化", error="未初始化")
+
+    actual_chat_id = client.open_window(chat_id=chat_id, notifier=_ToolNotifier(contact_id, chat_id or ""))
+    if chat_id is None:
+        client._notifiers[actual_chat_id] = _ToolNotifier(contact_id, actual_chat_id)
+
+    return ToolResult(title="窗口已打开", output=f"窗口 ID: {actual_chat_id}")
+
+
+@tool(
+    hidden_params=["context"],
+    display={
+        "zh": {"name": "关闭 IM 窗口", "params": {"contact_id": "Agent ID", "chat_id": "窗口 ID"}},
+        "en": {"name": "Close IM Window", "params": {"contact_id": "Agent ID", "chat_id": "Window ID"}},
+    }
+)
+async def im_close_window(
+    contact_id: str,
+    chat_id: str,
+    context: Optional[ToolContext] = None,
+) -> ToolResult:
+    """关闭一个聊天窗口。
+
+    Args:
+        contact_id: Agent ID
+        chat_id: 窗口 ID
+    """
+    client = _clients.get(contact_id)
+    if client is None:
+        return ToolResult(title="未连接", output=f"错误: {contact_id} 未初始化", error="未初始化")
+
+    client.close_window(chat_id)
+    _notifications.pop((contact_id, chat_id), None)
+    return ToolResult(title="窗口已关闭", output=f"已关闭窗口: {chat_id}")
+
+
+# ── Tool 3: 检查通知 ──
+
+@tool(
+    hidden_params=["context"],
+    display={
+        "zh": {"name": "检查 IM 新消息通知", "params": {"contact_id": "Agent ID", "chat_id": "窗口 ID"}},
+        "en": {"name": "Check IM Notifications", "params": {"contact_id": "Agent ID", "chat_id": "Window ID"}},
+    }
+)
+async def im_check_notification(
+    contact_id: str,
+    chat_id: str,
+    context: Optional[ToolContext] = None,
+) -> ToolResult:
+    """检查某个窗口是否有新消息通知。
+
+    Args:
+        contact_id: Agent ID
+        chat_id: 窗口 ID
+    """
+    notification = _notifications.pop((contact_id, chat_id), None)
+    if notification is None:
+        return ToolResult(title="无新消息", output="当前没有新消息通知")
+
+    return ToolResult(
+        title=f"有 {notification['count']} 条新消息",
+        output=json.dumps(notification, ensure_ascii=False),
+        metadata=notification,
+    )
+
+
+# ── Tool 4: 接收消息 ──
+
+@tool(
+    hidden_params=["context"],
+    display={
+        "zh": {"name": "接收 IM 消息", "params": {"contact_id": "Agent ID", "chat_id": "窗口 ID"}},
+        "en": {"name": "Receive IM Messages", "params": {"contact_id": "Agent ID", "chat_id": "Window ID"}},
+    }
+)
+async def im_receive_messages(
+    contact_id: str,
+    chat_id: str,
+    context: Optional[ToolContext] = None,
+) -> ToolResult:
+    """读取某个窗口的待处理消息。
+
+    Args:
+        contact_id: Agent ID
+        chat_id: 窗口 ID
+    """
+    client = _clients.get(contact_id)
+    if client is None:
+        return ToolResult(title="未连接", output=f"错误: {contact_id} 未初始化", error="未初始化")
+
+    raw = client.read_pending(chat_id)
+    if not raw:
+        return ToolResult(title="无待处理消息", output="[]")
+
+    messages = [
+        {
+            "sender": m.get("sender", "unknown"),
+            "sender_chat_id": m.get("sender_chat_id"),
+            "content": m.get("content", ""),
+            "msg_type": m.get("msg_type", "chat"),
+        }
+        for m in raw
+    ]
+    return ToolResult(
+        title=f"收到 {len(messages)} 条消息",
+        output=json.dumps(messages, ensure_ascii=False, indent=2),
+        metadata={"messages": messages},
+    )
+
+
+# ── Tool 5: 发送消息 ──
+
+@tool(
+    hidden_params=["context"],
+    display={
+        "zh": {"name": "发送 IM 消息", "params": {"contact_id": "Agent ID", "chat_id": "窗口 ID", "receiver": "接收者 ID", "content": "消息内容"}},
+        "en": {"name": "Send IM Message", "params": {"contact_id": "Agent ID", "chat_id": "Window ID", "receiver": "Receiver ID", "content": "Message content"}},
+    }
+)
+async def im_send_message(
+    contact_id: str,
+    chat_id: str,
+    receiver: str,
+    content: str,
+    msg_type: str = "chat",
+    receiver_chat_id: str | None = None,
+    context: Optional[ToolContext] = None,
+) -> ToolResult:
+    """从某个窗口发送消息。
+
+    Args:
+        contact_id: 发送方 Agent ID
+        chat_id: 发送方窗口 ID
+        receiver: 接收方 contact_id
+        content: 消息内容
+        msg_type: 消息类型
+        receiver_chat_id: 接收方窗口 ID（不指定则广播）
+    """
+    client = _clients.get(contact_id)
+    if client is None:
+        return ToolResult(title="未连接", output=f"错误: {contact_id} 未初始化", error="未初始化")
+
+    client.send_message(chat_id, receiver, content, msg_type, receiver_chat_id)
+    target = f"{receiver}:{receiver_chat_id}" if receiver_chat_id else f"{receiver}:*"
+    return ToolResult(
+        title=f"已发送给 {target}",
+        output=f"[{contact_id}:{chat_id}] 已发送给 {target}: {content[:50]}",
+    )
+
+
+# ── Tool 6: 查询联系人 ──
+
+@tool(
+    hidden_params=["context"],
+    display={
+        "zh": {"name": "查询 IM 联系人", "params": {"contact_id": "Agent ID", "server_http_url": "服务器 HTTP 地址"}},
+        "en": {"name": "Get IM Contacts", "params": {"contact_id": "Agent ID", "server_http_url": "Server HTTP URL"}},
+    }
+)
+async def im_get_contacts(
+    contact_id: str,
+    server_http_url: str = "http://localhost:8000",
+    context: Optional[ToolContext] = None,
+) -> ToolResult:
+    """查询联系人列表和当前在线用户。
+
+    Args:
+        contact_id: Agent ID
+        server_http_url: Server 的 HTTP 地址
+    """
+    if contact_id not in _clients:
+        return ToolResult(title="未连接", output=f"错误: {contact_id} 未初始化", error="未初始化")
+
+    import httpx
+    result = {}
+    async with httpx.AsyncClient() as http:
+        try:
+            r = await http.get(f"{server_http_url}/contacts/{contact_id}")
+            result["contacts"] = r.json().get("contacts", [])
+        except Exception as e:
+            result["contacts_error"] = str(e)
+        try:
+            r = await http.get(f"{server_http_url}/health")
+            result["online"] = r.json().get("online", {})
+        except Exception as e:
+            result["online_error"] = str(e)
+
+    return ToolResult(
+        title="联系人查询完成",
+        output=json.dumps(result, ensure_ascii=False, indent=2),
+        metadata=result,
+    )
+
+
+# ── Tool 7: 查询聊天历史 ──
+
+@tool(
+    hidden_params=["context"],
+    display={
+        "zh": {"name": "查询 IM 聊天历史", "params": {"contact_id": "Agent ID", "chat_id": "窗口 ID", "peer_id": "联系人 ID", "limit": "最大条数"}},
+        "en": {"name": "Get IM Chat History", "params": {"contact_id": "Agent ID", "chat_id": "Window ID", "peer_id": "Contact ID", "limit": "Max records"}},
+    }
+)
+async def im_get_chat_history(
+    contact_id: str,
+    chat_id: str,
+    peer_id: str | None = None,
+    limit: int = 20,
+    context: Optional[ToolContext] = None,
+) -> ToolResult:
+    """查询某个窗口的聊天历史。
+
+    Args:
+        contact_id: Agent ID
+        chat_id: 窗口 ID
+        peer_id: 筛选与某个联系人的聊天（留空返回所有）
+        limit: 最多返回条数
+    """
+    client = _clients.get(contact_id)
+    if client is None:
+        return ToolResult(title="未连接", output=f"错误: {contact_id} 未初始化", error="未初始化")
+
+    messages = client.get_chat_history(chat_id, peer_id, limit)
+    return ToolResult(
+        title=f"查到 {len(messages)} 条记录",
+        output=json.dumps(messages, ensure_ascii=False, indent=2),
+        metadata={"messages": messages},
+    )

agent/tools/builtin/resource.py

@@ -0,0 +1,64 @@
+"""
+资源查询工具 - KnowHub 工具表 API 封装
+"""
+
+import os
+import httpx
+from typing import List, Dict, Optional, Any
+from agent.tools import tool, ToolResult
+
+KNOWHUB_API = os.getenv("KNOWHUB_API", "http://43.106.118.91:9999").rstrip("/")
+
+
+@tool(    
+    description="列出知识库中的所有工具资源"
+)
+def resource_list_tools(
+    category: Optional[str] = None,
+) -> ToolResult:
+    """列出所有工具资源，可选按分类过滤
+
+    Args:
+        category: 可选的分类过滤，如 "plugin", "model" 等
+    """
+    try:
+        resp = httpx.get(f"{KNOWHUB_API}/api/resource", params={"limit": 1000}, timeout=60.0)
+        resp.raise_for_status()
+        data = resp.json()
+
+        # 处理返回格式
+        if isinstance(data, dict):
+            data = data.get("results") or data.get("data") or []
+
+        # 过滤工具
+        tools = [r for r in data if isinstance(r, dict) and r.get("id", "").startswith("tools/")]
+
+        # 按分类过滤
+        if category:
+            tools = [t for t in tools if t.get("metadata", {}).get("category") == category]
+
+        result = {
+            "total": len(tools),
+            "tools": [{"id": t["id"], "title": t.get("title", ""), "category": t.get("metadata", {}).get("category")} for t in tools]
+        }
+
+        return ToolResult(title="工具列表", output=str(result))
+    except Exception as e:
+        return ToolResult(title="查询失败", output="", error=str(e))
+
+
+@tool(
+    description="获取指定工具的详细信息",
+)
+def resource_get_tool(tool_id: str) -> ToolResult:
+    """获取工具详情
+
+    Args:
+        tool_id: 工具ID，如 "tools/image_gen/comfyui"
+    """
+    try:
+        resp = httpx.get(f"{KNOWHUB_API}/api/resource/{tool_id}", timeout=10.0)
+        resp.raise_for_status()
+        return ToolResult(title=f"工具详情: {tool_id}", output=str(resp.json()))
+    except Exception as e:
+        return ToolResult(title="获取工具详情失败", output="", error=str(e))

agent/tools/builtin/search.py

@@ -316,13 +316,18 @@ async def search_posts(
 
         # 拼接封面图网格
         images = []
-        collage_b64 = await _build_collage(posts)
-        if collage_b64:
-            images.append({
-                "type": "base64",
-                "media_type": "image/png",
-                "data": collage_b64
-            })
+        try:
+            collage_b64 = await _build_collage(posts)
+            if collage_b64:
+                images.append({
+                    "type": "base64",
+                    "media_type": "image/png",
+                    "data": collage_b64
+                })
+        except Exception as collage_error:
+            # 图片拼接失败不影响主流程，记录错误但继续返回结果
+            import logging
+            logging.warning(f"Failed to build collage for {channel_value}: {collage_error}")
 
         output_data = {
             "code": data.get("code"),

examples/restore/config.py

@@ -65,3 +65,10 @@ LOG_FILE = None  # 设置为文件路径可以同时输出到文件
 # 可选值: "cloud" (云浏览器) 或 "local" (本地浏览器) 或 "container" (容器浏览器，支持预配置账户)
 BROWSER_TYPE = "local"
 HEADLESS = False
+
+# ===== IM 配置 =====
+IM_ENABLED = True                          # 是否启动 IM Client
+IM_CONTACT_ID = "agent_restore"            # Agent 在 IM 系统中的身份 ID
+IM_SERVER_URL = "ws://localhost:8000"      # IM Server WebSocket 地址
+IM_WINDOW_MODE = True                      # 窗口模式（True=每次运行消息隔离，推荐）
+IM_NOTIFY_INTERVAL = 10.0                  # 新消息检查间隔（秒）

examples/restore/requirement.prompt

@@ -6,62 +6,39 @@ temperature: 0.3
 $system$
 
 ## 角色
-你是有空杯心态、擅长搜索调研SOTA工具和方案的社媒内容制作专家，擅长规划内容制作流程和计划。
-你的任务是，根据需求，充分调研最新实现方案，制定内容制作策略。尽可能使用AI工具或获取网络资源来完成内容制作，尽量减少实景拍摄。
+你是一个友好的 IM 沟通助手，能够通过即时消息与用户进行对话。
 
+## 可用工具
+你已连接到 IM 系统，可以使用以下工具：
+- `im_check_notification`：检查是否有新消息通知
+- `im_receive_messages`：读取待处理的新消息
+- `im_send_message`：发送消息给指定联系人
+- `im_get_contacts`：查询联系人列表和在线用户
+- `im_get_chat_history`：查询聊天历史
 
 ## 工作流程
 
-### 第一步：了解制作需求
-读取 `%input_dir%/analysis.json`，了解内容品类、主要特征，并提取亮点、下限点和需求清单。
+### 第一步：查看在线联系人
+调用 `im_get_contacts` 查看当前有哪些在线用户。
 
-### 第二步：以制定内容制作的流程和计划为目标，梳理调研需求
-根据制作需求，判断为了制定SOTA、可靠的制作工序规划，需要调研哪些方向的信息。
+### 第二步：主动打招呼
+选择一个在线联系人，用 `im_send_message` 发送一条友好的问候消息，介绍自己是 IM 沟通助手。
 
-### 第三步：循环迭代地向调研agent提问、评估调研结果、更新内容制作工序计划
-1. **提问**：向 subagent 提出调研问题
-   - MUST 调用工具 `agent(task="string - 一句话描述调研需求", agent_type="research")`
-   - **严格禁止**在 task 中预设猜想的具体工具名称或示例
-2. **评估**：subagent 返回后（可能是阶段性结果），读取调研结果并评估：
-   - **相关性**：找到的方案/工具是不是我要的方向？
-   - **可用性**：找到的工具能不能被 agent 使用？（过滤纯手机 app、本地桌面应用如 PS 等）
-   - **时效性**：找到的工具是不是过时了？（AI工具迭代很快，6个月前的信息都大概率过时了）
-   - **信息完整性**：找到的信息是否足够支撑后续选择？（信息够不够）
-3. **追问或结束**：
-   - 结论为"需补充"→ 用 `continue_from` 调用同一个 subagent，**在 task 中明确告知**：
-     - 还缺什么：缺少哪些必需/建议信息，或需要补充哪些方向的工具
-     - 建议搜索方向：给出具体的搜索建议（如"搜索该工具的用户评价"、"寻找该领域的其他工具"）
-   - 结论为"通过"→ 进入下一个问题或结束调研
-4. **基于最新信息思考**：
-   - 最新信息是否带来了新的思路？
-   - 是否需要更新原来的调研需求分析、提出新的调研问题？
-5. **创建或更新制作思路**
-   - 根据最新信息，撰写或更新制作思路（路径：%output_dir%/plan_thinking.md"）
-循环1-5的步骤，直到你对获取到的信息感到充分和满意。预期每个调研方向会经历 2-3 轮追问。
+### 第三步：进入消息循环
+反复执行以下步骤：
+1. 调用 `im_check_notification` 检查新消息
+2. 如果有新消息，调用 `im_receive_messages` 读取内容
+3. 根据消息内容思考并生成回复
+4. 调用 `im_send_message` 发送回复
+5. 如果没有新消息，等待片刻后继续检查
 
-### 第四步：制定制作工序计划
-综合所有调研结果和制作思路思考，确定最终的制作计划。
-
-**基于工具评估，选择合适工具**：
-1. **内在维度**（工具自带的属性）
-   - 时效性：越新越好
-   - 智能化：越智能的越好（如：AI 工具比非 AI 工具工具好）
-   - 通用性：越通用的越好
-2. **外部置信度**（外界的反馈与背书）
-   - 交叉验证（曝光率）：在不同平台、不同内容中提及次数越多的越好
-   - 专家/平台背书（权威性）：
-     - 赛道内头部 KOL 的推荐
-     - 专业平台的榜单（如 Hugging Face 榜单、liblib 热门榜单）
-   - 帖子本身热度高、评论正面反馈多
-   - 有实际效果案例展示
-
-**选择策略**：
-- 优先选择"内在维度强 + 外部置信度高"的工具
-- 如果工具在某个维度较弱，需要在"备选工具"中列出替代方案
-- 如果多个工具能力相近，选择外部置信度更高的
-
-**输出**：`%output_dir%/plan.md`，需要包含：主要步骤和各步骤的：理由、输入、输出、关联需求、风险（若有）、其他（若有）
+### 沟通原则
+- 用中文沟通
+- 回复要简洁、有帮助
+- 可以主动提问、引导对话
+- 如果对方问了你不确定的问题，诚实说明
+- 保持友好和耐心
 
 $user$
-基于以下输入目录中的需求分析，进行充分的调研，并在调研过程中持续思考、根据调研所得信息不断调整调研方向；最终制定指定内容的基于 SOTA 工具的内容制作计划：
-%input_dir%
+请开始工作：先查看在线联系人，主动向其中一位打招呼，然后进入消息循环持续对话。你等的久一点，别动不动就结束了。
+

examples/restore/requirement_0.prompt

@@ -0,0 +1,67 @@
+---
+model: qwen3.5-plus
+temperature: 0.3
+---
+
+$system$
+
+## 角色
+你是有空杯心态、擅长搜索调研SOTA工具和方案的社媒内容制作专家，擅长规划内容制作流程和计划。
+你的任务是，根据需求，充分调研最新实现方案，制定内容制作策略。尽可能使用AI工具或获取网络资源来完成内容制作，尽量减少实景拍摄。
+
+
+## 工作流程
+
+### 第一步：了解制作需求
+读取 `%input_dir%/analysis.json`，了解内容品类、主要特征，并提取亮点、下限点和需求清单。
+
+### 第二步：以制定内容制作的流程和计划为目标，梳理调研需求
+根据制作需求，判断为了制定SOTA、可靠的制作工序规划，需要调研哪些方向的信息。
+
+### 第三步：循环迭代地向调研agent提问、评估调研结果、更新内容制作工序计划
+1. **提问**：向 subagent 提出调研问题
+   - MUST 调用工具 `agent(task="string - 一句话描述调研需求", agent_type="research")`
+   - **严格禁止**在 task 中预设猜想的具体工具名称或示例
+2. **评估**：subagent 返回后（可能是阶段性结果），读取调研结果并评估：
+   - **相关性**：找到的方案/工具是不是我要的方向？
+   - **可用性**：找到的工具能不能被 agent 使用？（过滤纯手机 app、本地桌面应用如 PS 等）
+   - **时效性**：找到的工具是不是过时了？（AI工具迭代很快，6个月前的信息都大概率过时了）
+   - **信息完整性**：找到的信息是否足够支撑后续选择？（信息够不够）
+3. **追问或结束**：
+   - 结论为"需补充"→ 用 `continue_from` 调用同一个 subagent，**在 task 中明确告知**：
+     - 还缺什么：缺少哪些必需/建议信息，或需要补充哪些方向的工具
+     - 建议搜索方向：给出具体的搜索建议（如"搜索该工具的用户评价"、"寻找该领域的其他工具"）
+   - 结论为"通过"→ 进入下一个问题或结束调研
+4. **基于最新信息思考**：
+   - 最新信息是否带来了新的思路？
+   - 是否需要更新原来的调研需求分析、提出新的调研问题？
+5. **创建或更新制作思路**
+   - 根据最新信息，撰写或更新制作思路（路径：%output_dir%/plan_thinking.md"）
+循环1-5的步骤，直到你对获取到的信息感到充分和满意。预期每个调研方向会经历 2-3 轮追问。
+
+### 第四步：制定制作工序计划
+综合所有调研结果和制作思路思考，确定最终的制作计划。
+
+**基于工具评估，选择合适工具**：
+1. **内在维度**（工具自带的属性）
+   - 时效性：越新越好
+   - 智能化：越智能的越好（如：AI 工具比非 AI 工具工具好）
+   - 通用性：越通用的越好
+2. **外部置信度**（外界的反馈与背书）
+   - 交叉验证（曝光率）：在不同平台、不同内容中提及次数越多的越好
+   - 专家/平台背书（权威性）：
+     - 赛道内头部 KOL 的推荐
+     - 专业平台的榜单（如 Hugging Face 榜单、liblib 热门榜单）
+   - 帖子本身热度高、评论正面反馈多
+   - 有实际效果案例展示
+
+**选择策略**：
+- 优先选择"内在维度强 + 外部置信度高"的工具
+- 如果工具在某个维度较弱，需要在"备选工具"中列出替代方案
+- 如果多个工具能力相近，选择外部置信度更高的
+
+**输出**：`%output_dir%/plan.md`，需要包含：主要步骤和各步骤的：理由、输入、输出、关联需求、风险（若有）、其他（若有）
+
+$user$
+基于以下输入目录中的需求分析，进行充分的调研，并在调研过程中持续思考、根据调研所得信息不断调整调研方向；最终制定指定内容的基于 SOTA 工具的内容制作计划：
+%input_dir%

examples/restore/research.prompt

@@ -20,7 +20,36 @@ $system$
 1. **知识库优先**：用 `knowledge_search` 按需求关键词搜索，查看已有策略经验、工具评估、工作流总结
 2. **线上调研**：知识库结果不充分时，进行线上搜索
 
-**搜索方法**：
+**可用的搜索工具**：
+
+1. **search_posts** - 中文内容平台搜索（用户体验、案例、教程）
+   - 小红书 (xhs) - 用户真实体验、产品评价、使用技巧
+   - 微信公众号 (gzh) - 深度文章、行业分析、技术教程
+   - 知乎 (zhihu) - 专业问答、技术讨论、方案对比
+   - B站 (bili) - 视频教程、实操演示、工作流展示
+   - 抖音 (douyin) - 短视频内容、快速概览
+   - 视频号 (sph) - 微信视频内容
+   - 微博 (weibo) - 热点讨论、快讯
+   - 头条 (toutiao) - 资讯文章
+
+2. **youtube_search** + **youtube_detail** - YouTube 视频搜索
+   - 国际教程、官方演示、技术讲解
+   - `youtube_detail` 可获取视频字幕（默认包含）
+
+3. **web_search** - 通用网页搜索
+   - 官网、文档、GitHub、技术博客
+
+4. **browser_use** - 浏览器自动化（当其他工具不够用时）
+
+**搜索策略**：
+
+- **渠道选择原则**：
+  - 找用户体验/真实案例 → 小红书、知乎
+  - 找深度分析/技术方案 → 微信公众号、知乎
+  - 找视频教程/操作演示 → YouTube、B站
+  - 找官网/文档/代码 → web_search
+  - 找国际内容 → YouTube、web_search
+  - 找中文内容 → search_posts (优先小红书、公众号、知乎)
 
 - **需求驱动，不预设工具**：从需求出发构建 query，从结果中发现工具
   - **query 构建原则**：从需求出发，不要预设工具
@@ -35,6 +64,8 @@ $system$
 
 - **粗到细**：先找该类型下有哪些工具/方案，再对相关的深入调研
 
+- **多渠道验证**：同一个工具/方案，尽量在多个渠道验证（如：小红书看用户评价 + 公众号看深度分析 + YouTube 看实操演示）
+
 ### 第三步：反思与调整
 
 在搜索过程中，你需要主动进行反思和调整：
@@ -112,5 +143,7 @@ $system$
 
 
 ## 注意事项
-- `search_posts` 不好用时改用 `browser-use`
+- 优先使用 `search_posts` (小红书、公众号、知乎) 和 `youtube_search` 进行调研
+- 这些平台的数据质量高，用户体验真实，是调研的主要来源
+- `search_posts` 不好用时改用 `browser_use`
 - 如果调研过程中遇到不确定的问题，要停下来询问用户

examples/restore/run.py

@@ -41,10 +41,11 @@ from agent.utils import setup_logging
 from agent.tools.builtin.browser.baseClass import init_browser_session, kill_browser_session
 
 # 导入自定义工具（触发 @tool 注册）
-from .tools.reflect import reflect
+# from .tools.reflect import reflect
 
 # 导入项目配置
 from config import RUN_CONFIG, SKILLS_DIR, TRACE_STORE_PATH, DEBUG, LOG_LEVEL, LOG_FILE, BROWSER_TYPE, HEADLESS, INPUT_DIR, OUTPUT_DIR
+from config import IM_ENABLED, IM_CONTACT_ID, IM_SERVER_URL, IM_WINDOW_MODE, IM_NOTIFY_INTERVAL
 
 
 async def main():
@@ -98,6 +99,20 @@ async def main():
     )
     print(f"   ✅ {browser_mode_name}初始化完成\n")
 
+    # 5.5 初始化 IM Client（可选）
+    if IM_ENABLED:
+        from agent.tools.builtin.im.chat import im_setup
+        print("5.5 初始化 IM Client...")
+        mode_name = "窗口模式" if IM_WINDOW_MODE else "普通模式"
+        print(f"   - 身份: {IM_CONTACT_ID}, 服务器: {IM_SERVER_URL}, {mode_name}")
+        result = await im_setup(
+            contact_id=IM_CONTACT_ID,
+            server_url=IM_SERVER_URL,
+            window_mode=IM_WINDOW_MODE,
+            notify_interval=IM_NOTIFY_INTERVAL,
+        )
+        print(f"   ✅ {result.output}\n")
+
     # 6. 创建 Agent Runner
     print("6. 创建 Agent Runner...")
     print(f"   - Skills 目录: {SKILLS_DIR}")

im-client/AGENT_GUIDE.md

@@ -0,0 +1,263 @@
+# Agent 接入 IM 系统指南
+
+本文档面向 Agent 开发者（包括 Claude Code），说明如何为 Agent 接入 IM 通信能力。
+
+---
+
+## 前置条件
+
+1. IM Server 已启动: `cd im-server && uvicorn main:app --port 8000`
+2. 安装依赖: `pip install websockets pydantic filelock`
+
+---
+
+## 核心概念
+
+| 概念 | 说明 |
+|---|---|
+| `contact_id` | Agent 在 IM 系统中的唯一身份标识，如 `"agent_alice"` |
+| `IMClient` | 长驻 asyncio 服务，负责 WebSocket 连接和文件读写 |
+| `chat_id` | 窗口模式下的会话隔离 ID，每次 Agent 运行可生成新的 |
+| Server | 纯转发，只管路由，不存消息 |
+
+---
+
+## 快速接入（3 步）
+
+### 第 1 步：启动 Client
+
+```python
+import asyncio
+import sys
+sys.path.insert(0, "/path/to/IM-Server/im-client")
+
+from client import IMClient
+
+client = IMClient(
+    contact_id="my_agent",          # 你的 Agent ID
+    server_url="ws://localhost:8000" # Server 地址
+)
+
+# 后台启动（在 Agent 的 asyncio 事件循环中）
+asyncio.create_task(client.run())
+```
+
+### 第 2 步：发消息
+
+```python
+# 给 contact_id 为 "bob" 的联系人发消息
+client.send_message(receiver="bob", content="你好，我是 my_agent")
+
+# 发送图片（用 URL）
+client.send_message(receiver="bob", content="https://example.com/img.png", msg_type="image")
+```
+
+消息会立即进入发送队列，由 Client 通过 WebSocket 发出��
+
+### 第 3 步：收消息
+
+```python
+# 读取并清空待处理消息
+messages = client.read_pending()
+for msg in messages:
+    sender = msg["sender"]       # 谁发的
+    content = msg["content"]     # 消息内容
+    msg_type = msg["msg_type"]   # chat | image | video
+    print(f"{sender}: {content}")
+```
+
+调用 `read_pending()` 后，`in_pending.json` 会被清空。消息已永久记录在 `chatbox.jsonl` 中。
+
+---
+
+## 查看联系人
+
+Server 维护联系人表，通过 HTTP API 查询：
+
+```bash
+# 查看某用户的联系人
+curl http://localhost:8000/contacts/my_agent
+
+# 添加联系人
+curl -X POST "http://localhost:8000/contacts/my_agent/add?contact_id=bob"
+
+# 查看谁在线
+curl http://localhost:8000/health
+```
+
+Agent 可以通过 HTTP 请求查询联系人列表，决定给谁发消息。所有在 Server 联系人表中的用户，只要在线就可以互相通信。
+
+---
+
+## 窗口模式（推荐 Agent 使用）
+
+每次 Agent 运行时开启新窗口，避免被上一次的历史消息干扰：
+
+```python
+client = IMClient(
+    contact_id="my_agent",
+    server_url="ws://localhost:8000",
+    window_mode=True  # 自动生成新 chat_id
+)
+# client.chat_id => "20260326_082445_a3f9e1"
+```
+
+恢复上次窗口（如果需要）：
+
+```python
+client = IMClient(
+    contact_id="my_agent",
+    window_mode=True,
+    chat_id="20260326_082445_a3f9e1"  # 指定旧窗口
+)
+```
+
+### 窗口模式 vs 普通模式的区别
+
+| | 普通模式 | 窗口模式 |
+|---|---|---|
+| 数据目录 | `data/{contact_id}/` | `data/{contact_id}/windows/{chat_id}/` |
+| 消息隔离 | 所有消息在一起 | 每次运行独立 |
+| 适用场景 | 人类用户、长期运行 | Agent、每次任务独立 |
+
+---
+
+## 自定义通知
+
+当有新消息到达时，Client 会定期调用通知回调。Agent 可以自定义处理方式：
+
+```python
+from notifier import AgentNotifier
+
+class MyAgentNotifier(AgentNotifier):
+    async def notify(self, count: int, from_contacts: list[str]):
+        # count: 新消息条数
+        # from_contacts: 发送者的 contact_id 列表
+        print(f"收到 {count} 条消息，来自 {from_contacts}")
+        # 在这里触发 Agent 的处理逻辑，例如：
+        # - 调用 Agent 的 tool
+        # - 写入 Agent 的任务队列
+        # - 中断当前任务去处理消息
+
+client = IMClient(
+    contact_id="my_agent",
+    notifier=MyAgentNotifier(),
+    notify_interval=10.0  # 每 10 秒检查一次（默认 30 秒）
+)
+```
+
+---
+
+## 文件结构
+
+Client 的所有数据存储在本地文件中：
+
+```
+data/{contact_id}/
+├── chatbox.jsonl          # 所有消息历史（一行一条 JSON）
+├── in_pending.json        # 待处理的收到消息（JSON 数组）
+├── out_pending.jsonl      # 发送失败的消息
+└── windows/               # 窗口模式
+    └── {chat_id}/
+        ├── chatbox.jsonl
+        ├── in_pending.json
+        └── out_pending.jsonl
+```
+
+### chatbox.jsonl 格式
+
+每行一条 JSON，收发消息都在这里：
+
+```jsonl
+{"msg_id":"a1b2c3","sender":"my_agent","receiver":"bob","content":"你好","msg_type":"chat"}
+{"msg_id":"d4e5f6","sender":"bob","receiver":"my_agent","content":"你好！","msg_type":"chat"}
+```
+
+### in_pending.json 格式
+
+JSON 数组，`read_pending()` 调用后清空：
+
+```json
+[
+  {"msg_id":"d4e5f6","sender":"bob","receiver":"my_agent","content":"你好！","msg_type":"chat"}
+]
+```
+
+### out_pending.jsonl 格式
+
+仅记录发送失败的消息（对方不在线、网络断开等），每行一条：
+
+```jsonl
+{"msg_id":"g7h8i9","sender":"my_agent","receiver":"charlie","content":"在吗？","msg_type":"chat"}
+```
+
+---
+
+## 消息协议
+
+所有消息使用统一的 JSON 格式：
+
+```json
+{
+    "msg_id": "a1b2c3d4e5f6",
+    "sender": "my_agent",
+    "receiver": "bob",
+    "content": "消息内容 / 图片URL / 视频URL",
+    "msg_type": "chat"
+}
+```
+
+`msg_type` 可选值：`chat`（文本）、`image`、`video`、`system`。
+
+媒体消息的 `content` 字段填 URL。
+
+---
+
+## 完整示例：Agent 接入
+
+```python
+import asyncio
+import sys
+sys.path.insert(0, "/path/to/IM-Server/im-client")
+
+from client import IMClient
+from notifier import AgentNotifier
+
+
+class MyNotifier(AgentNotifier):
+    def __init__(self, client_ref):
+        self._client_ref = client_ref
+
+    async def notify(self, count, from_contacts):
+        print(f"[IM] {count} 条新消息，来自 {from_contacts}")
+        messages = self._client_ref.read_pending()
+        for msg in messages:
+            # 在这里处理消息
+            print(f"  {msg['sender']}: {msg['content']}")
+
+
+async def main():
+    client = IMClient(
+        contact_id="my_agent",
+        server_url="ws://localhost:8000",
+        window_mode=True,
+        notify_interval=10.0,
+    )
+    client.notifier = MyNotifier(client)
+
+    # 后台启动 client
+    client_task = asyncio.create_task(client.run())
+
+    # 模拟 Agent 工作
+    await asyncio.sleep(2)  # 等待连接建立
+    client.send_message("bob", "你好 Bob，我上线了！")
+
+    # Agent 继续做自己的事...
+    await asyncio.sleep(60)
+
+    client_task.cancel()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
+```

im-client/client.py

@@ -0,0 +1,308 @@
+import asyncio
+import json
+import logging
+import os
+import tempfile
+import uuid
+from datetime import datetime
+from pathlib import Path
+
+import websockets
+from filelock import FileLock
+
+from protocol import IMMessage, IMResponse
+from notifier import AgentNotifier, ConsoleNotifier
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s [CLIENT:%(name)s] %(message)s")
+
+
+class ChatWindow:
+    """单个聊天窗口的数据管理。"""
+
+    def __init__(self, chat_id: str, data_dir: Path):
+        self.chat_id = chat_id
+        self.data_dir = data_dir
+        self.data_dir.mkdir(parents=True, exist_ok=True)
+
+        self.chatbox_path = data_dir / "chatbox.jsonl"
+        self.in_pending_path = data_dir / "in_pending.json"
+        self.out_pending_path = data_dir / "out_pending.jsonl"
+
+        # 文件锁
+        self._in_pending_lock = FileLock(str(data_dir / ".in_pending.lock"))
+        self._out_pending_lock = FileLock(str(data_dir / ".out_pending.lock"))
+        self._chatbox_lock = FileLock(str(data_dir / ".chatbox.lock"))
+
+        # 初始化文件
+        if not self.chatbox_path.exists():
+            self.chatbox_path.write_text("")
+        if not self.in_pending_path.exists():
+            self.in_pending_path.write_text("[]")
+        if not self.out_pending_path.exists():
+            self.out_pending_path.write_text("")
+
+    def append_to_in_pending(self, msg: dict):
+        with self._in_pending_lock:
+            pending = self._load_json_array(self.in_pending_path)
+            pending.append(msg)
+            self._atomic_write_json(self.in_pending_path, pending)
+
+    def read_in_pending(self) -> list[dict]:
+        with self._in_pending_lock:
+            return self._load_json_array(self.in_pending_path)
+
+    def clear_in_pending(self):
+        with self._in_pending_lock:
+            self._atomic_write_json(self.in_pending_path, [])
+
+    def append_to_chatbox(self, msg: dict):
+        with self._chatbox_lock:
+            with open(self.chatbox_path, "a", encoding="utf-8") as f:
+                f.write(json.dumps(msg, ensure_ascii=False) + "\n")
+
+    def append_to_out_pending(self, msg: dict):
+        with self._out_pending_lock:
+            with open(self.out_pending_path, "a", encoding="utf-8") as f:
+                f.write(json.dumps(msg, ensure_ascii=False) + "\n")
+
+    @staticmethod
+    def _load_json_array(path: Path) -> list:
+        if not path.exists():
+            return []
+        text = path.read_text(encoding="utf-8").strip()
+        if not text:
+            return []
+        try:
+            data = json.loads(text)
+            return data if isinstance(data, list) else []
+        except json.JSONDecodeError:
+            return []
+
+    @staticmethod
+    def _atomic_write_json(path: Path, data):
+        tmp_fd, tmp_path = tempfile.mkstemp(dir=str(path.parent), suffix=".tmp")
+        try:
+            with os.fdopen(tmp_fd, "w", encoding="utf-8") as f:
+                json.dump(data, f, ensure_ascii=False, indent=2)
+            os.replace(tmp_path, str(path))
+        except Exception:
+            if os.path.exists(tmp_path):
+                os.unlink(tmp_path)
+            raise
+
+
+class IMClient:
+    """IM Client - 一个实例管理多个聊天窗口。
+
+    一个 Agent (contact_id) 对应一个 IMClient 实例。
+    该实例可以管理多个 chat_id（窗口），每个窗口有独立的消息存储。
+    """
+
+    def __init__(
+        self,
+        contact_id: str,
+        server_url: str = "ws://localhost:8000",
+        data_dir: str | None = None,
+        notify_interval: float = 30.0,
+    ):
+        self.contact_id = contact_id
+        self.server_url = server_url
+        self.notify_interval = notify_interval
+
+        self.base_dir = Path(data_dir) if data_dir else Path("data") / contact_id
+        self.base_dir.mkdir(parents=True, exist_ok=True)
+
+        # 窗口管理
+        self._windows: dict[str, ChatWindow] = {}
+        self._notifiers: dict[str, AgentNotifier] = {}
+
+        self.ws = None
+        self.log = logging.getLogger(contact_id)
+        self._send_queue = asyncio.Queue()
+
+    def open_window(self, chat_id: str | None = None, notifier: AgentNotifier | None = None) -> str:
+        """打开一个新窗口。
+
+        Args:
+            chat_id: 窗口 ID（留空自动生成）
+            notifier: 该窗口的通知器
+
+        Returns:
+            窗口的 chat_id
+        """
+        if chat_id is None:
+            chat_id = datetime.now().strftime("%Y%m%d_%H%M%S_") + uuid.uuid4().hex[:6]
+
+        if chat_id in self._windows:
+            return chat_id
+
+        window_dir = self.base_dir / "windows" / chat_id
+        self._windows[chat_id] = ChatWindow(chat_id, window_dir)
+        self._notifiers[chat_id] = notifier or ConsoleNotifier()
+
+        self.log.info(f"打开窗口: {chat_id}")
+        return chat_id
+
+    def close_window(self, chat_id: str):
+        """关闭一个窗口。"""
+        self._windows.pop(chat_id, None)
+        self._notifiers.pop(chat_id, None)
+        self.log.info(f"关闭窗口: {chat_id}")
+
+    def list_windows(self) -> list[str]:
+        """列出所有打开的窗口。"""
+        return list(self._windows.keys())
+
+    async def run(self):
+        """启动 Client 服务，自动重连。"""
+        while True:
+            try:
+                # 连接时不带 chat_id，因为一个实例管理多个窗口
+                ws_url = f"{self.server_url}/ws?contact_id={self.contact_id}&chat_id=__multi__"
+                self.log.info(f"连接 {ws_url} ...")
+                async with websockets.connect(ws_url) as ws:
+                    self.ws = ws
+                    self.log.info("已连接")
+                    await asyncio.gather(
+                        self._ws_listener(),
+                        self._send_worker(),
+                        self._pending_notifier(),
+                    )
+            except (websockets.ConnectionClosed, ConnectionRefusedError, OSError) as e:
+                self.log.warning(f"连接断开: {e}, 5 秒后重连...")
+                self.ws = None
+                await asyncio.sleep(5)
+            except asyncio.CancelledError:
+                self.log.info("服务停止")
+                break
+
+    async def _ws_listener(self):
+        """监听 WebSocket，根据 receiver_chat_id 分发到对应窗口。"""
+        async for raw in self.ws:
+            try:
+                data = json.loads(raw)
+            except json.JSONDecodeError:
+                self.log.warning(f"收到无效 JSON: {raw}")
+                continue
+
+            if "sender" in data and "receiver" in data:
+                # 聊天消息
+                receiver_chat_id = data.get("receiver_chat_id")
+
+                if receiver_chat_id and receiver_chat_id in self._windows:
+                    # 定向发送到指定窗口
+                    window = self._windows[receiver_chat_id]
+                    window.append_to_in_pending(data)
+                    window.append_to_chatbox(data)
+                    self.log.info(f"收到消息 -> 窗口 {receiver_chat_id}: {data['sender']}")
+                elif not receiver_chat_id:
+                    # 广播到所有窗口
+                    for chat_id, window in self._windows.items():
+                        window.append_to_in_pending(data)
+                        window.append_to_chatbox(data)
+                    self.log.info(f"收到消息 -> 广播到 {len(self._windows)} 个窗口: {data['sender']}")
+                else:
+                    self.log.warning(f"收到消息但窗口 {receiver_chat_id} 不存在")
+
+            elif "status" in data:
+                # 发送回执
+                resp = IMResponse(**data)
+                if resp.status == "success":
+                    self.log.info(f"消息 {resp.msg_id} 发送成功")
+                else:
+                    self.log.warning(f"消息 {resp.msg_id} 发送失败: {resp.error}")
+
+    async def _send_worker(self):
+        """从队列取消息并发送。"""
+        while True:
+            msg_data = await self._send_queue.get()
+            msg = IMMessage(sender=self.contact_id, **msg_data)
+            try:
+                await self.ws.send(msg.model_dump_json())
+                self.log.info(f"发送消息: -> {msg.receiver}:{msg.receiver_chat_id or '*'}")
+                # 记录到发送方窗口的 chatbox
+                if msg.sender_chat_id and msg.sender_chat_id in self._windows:
+                    self._windows[msg.sender_chat_id].append_to_chatbox(msg.model_dump())
+            except Exception as e:
+                self.log.error(f"发送失败: {e}")
+                if msg.sender_chat_id and msg.sender_chat_id in self._windows:
+                    self._windows[msg.sender_chat_id].append_to_out_pending(msg.model_dump())
+
+    async def _pending_notifier(self):
+        """轮询各窗口的 in_pending，有新消息就调通知回调。"""
+        while True:
+            for chat_id, window in list(self._windows.items()):
+                pending = window.read_in_pending()
+                if pending:
+                    senders = list(set(m.get("sender", "unknown") for m in pending))
+                    count = len(pending)
+                    notifier = self._notifiers.get(chat_id)
+                    if notifier:
+                        try:
+                            await notifier.notify(count=count, from_contacts=senders)
+                        except Exception as e:
+                            self.log.error(f"窗口 {chat_id} 通知回调异常: {e}")
+            await asyncio.sleep(self.notify_interval)
+
+    # ── Agent 调用的工具方法 ──
+
+    def read_pending(self, chat_id: str) -> list[dict]:
+        """读取某个窗口的待处理消息，并清空。"""
+        window = self._windows.get(chat_id)
+        if window is None:
+            return []
+        pending = window.read_in_pending()
+        if pending:
+            window.clear_in_pending()
+        return pending
+
+    def send_message(
+        self,
+        chat_id: str,
+        receiver: str,
+        content: str,
+        msg_type: str = "chat",
+        receiver_chat_id: str | None = None,
+    ):
+        """从某个窗口发送消息。"""
+        msg_data = {
+            "sender_chat_id": chat_id,
+            "receiver": receiver,
+            "content": content,
+            "msg_type": msg_type,
+            "receiver_chat_id": receiver_chat_id,
+        }
+        self._send_queue.put_nowait(msg_data)
+
+    def get_chat_history(self, chat_id: str, peer_id: str | None = None, limit: int = 20) -> list[dict]:
+        """查询某个窗口的聊天历史。"""
+        window = self._windows.get(chat_id)
+        if window is None or not window.chatbox_path.exists():
+            return []
+
+        lines = window.chatbox_path.read_text(encoding="utf-8").strip().splitlines()
+        messages = []
+        for line in reversed(lines):
+            if not line.strip():
+                continue
+            try:
+                m = json.loads(line)
+            except json.JSONDecodeError:
+                continue
+
+            if peer_id and m.get("sender") != peer_id and m.get("receiver") != peer_id:
+                continue
+
+            messages.append({
+                "sender": m.get("sender", "unknown"),
+                "receiver": m.get("receiver", "unknown"),
+                "content": m.get("content", ""),
+                "msg_type": m.get("msg_type", "chat"),
+            })
+
+            if len(messages) >= limit:
+                break
+
+        messages.reverse()
+        return messages
+

im-client/client.py.bak

@@ -0,0 +1,248 @@
+import asyncio
+import json
+import logging
+import os
+import tempfile
+import uuid
+from datetime import datetime
+from pathlib import Path
+
+import websockets
+from filelock import FileLock
+
+from protocol import IMMessage, IMResponse
+from notifier import AgentNotifier, ConsoleNotifier
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s [CLIENT:%(name)s] %(message)s")
+
+
+class IMClient:
+    """IM Client 长驻服务。
+
+    通过 WebSocket 连接 Server，通过文件与 Agent 交互。
+
+    文件约定 (data/{contact_id}/):
+        chatbox.jsonl     — 所有消息历史（收发都记录）
+        in_pending.json   — 收到的待处理消息 (JSON 数组)
+        out_pending.jsonl — 发送失败的消息
+
+    窗口模式 (window_mode=True):
+        每次运行生成新的 chat_id，消息按 chat_id 隔离
+        文件结构变为: data/{contact_id}/{chat_id}/...
+    """
+
+    def __init__(
+        self,
+        contact_id: str,
+        server_url: str = "ws://localhost:8000",
+        data_dir: str | None = None,
+        notifier: AgentNotifier | None = None,
+        notify_interval: float = 30.0,
+        window_mode: bool = False,
+        chat_id: str | None = None,
+    ):
+        self.contact_id = contact_id
+        self.server_url = server_url
+        self.notifier = notifier or ConsoleNotifier()
+        self.notify_interval = notify_interval
+        self.window_mode = window_mode
+
+        # 窗口模式：生成或使用指定的 chat_id
+        base_dir = Path(data_dir) if data_dir else Path("data") / contact_id
+        if window_mode:
+            self.chat_id = chat_id or datetime.now().strftime("%Y%m%d_%H%M%S_") + uuid.uuid4().hex[:6]
+            self.data_dir = base_dir / "windows" / self.chat_id
+        else:
+            self.chat_id = None
+            self.data_dir = base_dir
+
+        self.data_dir.mkdir(parents=True, exist_ok=True)
+
+        self.chatbox_path = self.data_dir / "chatbox.jsonl"
+        self.in_pending_path = self.data_dir / "in_pending.json"
+        self.out_pending_path = self.data_dir / "out_pending.jsonl"
+
+        # 文件锁
+        self._in_pending_lock = FileLock(str(self.data_dir / ".in_pending.lock"))
+        self._out_pending_lock = FileLock(str(self.data_dir / ".out_pending.lock"))
+        self._chatbox_lock = FileLock(str(self.data_dir / ".chatbox.lock"))
+
+        self.ws = None
+        self.log = logging.getLogger(f"{contact_id}:{self.chat_id}" if self.chat_id else contact_id)
+        self._send_queue = asyncio.Queue()
+
+        # 初始化文件
+        if not self.chatbox_path.exists():
+            self.chatbox_path.write_text("")
+        if not self.in_pending_path.exists():
+            self.in_pending_path.write_text("[]")
+        if not self.out_pending_path.exists():
+            self.out_pending_path.write_text("")
+
+    async def run(self):
+        """启动 Client 服务，自动重连。"""
+        while True:
+            try:
+                # 构造 WebSocket URL，带上 chat_id 参数
+                chat_id_param = self.chat_id or "default"
+                ws_url = f"{self.server_url}/ws?contact_id={self.contact_id}&chat_id={chat_id_param}"
+                self.log.info(f"连接 {ws_url} ...")
+                async with websockets.connect(ws_url) as ws:
+                    self.ws = ws
+                    self.log.info("已连接")
+                    await asyncio.gather(
+                        self._ws_listener(),
+                        self._send_worker(),
+                        self._pending_notifier(),
+                    )
+            except (websockets.ConnectionClosed, ConnectionRefusedError, OSError) as e:
+                self.log.warning(f"连接断开: {e}, 5 秒后重连...")
+                self.ws = None
+                await asyncio.sleep(5)
+            except asyncio.CancelledError:
+                self.log.info("服务停止")
+                break
+
+    # ── 协程 1: WebSocket 收消息 ──
+
+    async def _ws_listener(self):
+        """监听 WebSocket，聊天消息写 in_pending 和 chatbox，回执打日志。"""
+        async for raw in self.ws:
+            try:
+                data = json.loads(raw)
+            except json.JSONDecodeError:
+                self.log.warning(f"收到无效 JSON: {raw}")
+                continue
+
+            if "sender" in data and "receiver" in data:
+                # 聊天消息
+                self.log.info(f"收到消息: {data['sender']} -> {data['content'][:50]}")
+                self._append_to_in_pending(data)
+                self._append_to_chatbox(data)
+            elif "status" in data:
+                # 发送回执
+                resp = IMResponse(**data)
+                if resp.status == "success":
+                    self.log.info(f"消息 {resp.msg_id} 发送成功")
+                else:
+                    self.log.warning(f"消息 {resp.msg_id} 发送失败: {resp.error}")
+
+    # ── 协程 2: 发送队列处理 ──
+
+    async def _send_worker(self):
+        """从队列取消息并发送，失败则写入 out_pending。"""
+        while True:
+            msg_data = await self._send_queue.get()
+            # 填充 sender_chat_id
+            msg = IMMessage(
+                sender=self.contact_id,
+                sender_chat_id=self.chat_id or "default",
+                **msg_data
+            )
+            try:
+                await self.ws.send(msg.model_dump_json())
+                self.log.info(f"发送消息: -> {msg.receiver}:{msg.receiver_chat_id or '*'}")
+                # 记录到 chatbox
+                self._append_to_chatbox(msg.model_dump())
+            except Exception as e:
+                self.log.error(f"发送失败: {e}")
+                # 写入 out_pending
+                self._append_to_out_pending(msg.model_dump())
+
+    # ── 协程 3: 轮询 in_pending 通知 Agent ──
+
+    async def _pending_notifier(self):
+        """轮询 in_pending.json，有新消息就调通知回调。"""
+        while True:
+            pending = self._read_in_pending()
+            if pending:
+                senders = list(set(m.get("sender", "unknown") for m in pending))
+                count = len(pending)
+                try:
+                    await self.notifier.notify(count=count, from_contacts=senders)
+                except Exception as e:
+                    self.log.error(f"通知回调异常: {e}")
+            await asyncio.sleep(self.notify_interval)
+
+    # ── 文件操作 (原子性) ──
+
+    def _append_to_in_pending(self, msg: dict):
+        """将收到的消息追加到 in_pending.json。"""
+        with self._in_pending_lock:
+            pending = self._load_json_array(self.in_pending_path)
+            pending.append(msg)
+            self._atomic_write_json(self.in_pending_path, pending)
+
+    def _read_in_pending(self) -> list[dict]:
+        """读取 in_pending.json (不清空)。"""
+        with self._in_pending_lock:
+            return self._load_json_array(self.in_pending_path)
+
+    def _append_to_chatbox(self, msg: dict):
+        """追加消息到 chatbox.jsonl。"""
+        with self._chatbox_lock:
+            with open(self.chatbox_path, "a", encoding="utf-8") as f:
+                f.write(json.dumps(msg, ensure_ascii=False) + "\n")
+
+    def _append_to_out_pending(self, msg: dict):
+        """追加发送失败的消息到 out_pending.jsonl。"""
+        with self._out_pending_lock:
+            with open(self.out_pending_path, "a", encoding="utf-8") as f:
+                f.write(json.dumps(msg, ensure_ascii=False) + "\n")
+
+    # ── Agent 调用的工具方法 ──
+
+    def read_pending(self) -> list[dict]:
+        """Agent 读取 in_pending 中的消息，并清空。"""
+        with self._in_pending_lock:
+            pending = self._load_json_array(self.in_pending_path)
+            if not pending:
+                return []
+            # 清空 in_pending
+            self._atomic_write_json(self.in_pending_path, [])
+            return pending
+
+    def send_message(self, receiver: str, content: str, msg_type: str = "chat", receiver_chat_id: str | None = None):
+        """Agent 调用：将消息放入发送队列。
+
+        Args:
+            receiver: 接收方 contact_id
+            content: 消息内容
+            msg_type: 消息类型
+            receiver_chat_id: 接收方窗口 ID（指定则定向发送，否则广播给该 contact_id 的所有窗口）
+        """
+        msg_data = {
+            "receiver": receiver,
+            "content": content,
+            "msg_type": msg_type,
+            "receiver_chat_id": receiver_chat_id
+        }
+        self._send_queue.put_nowait(msg_data)
+
+    # ── 工具方法 ──
+
+    @staticmethod
+    def _load_json_array(path: Path) -> list:
+        if not path.exists():
+            return []
+        text = path.read_text(encoding="utf-8").strip()
+        if not text:
+            return []
+        try:
+            data = json.loads(text)
+            return data if isinstance(data, list) else []
+        except json.JSONDecodeError:
+            return []
+
+    @staticmethod
+    def _atomic_write_json(path: Path, data):
+        """原子写入：先写临时文件再 rename。"""
+        tmp_fd, tmp_path = tempfile.mkstemp(dir=str(path.parent), suffix=".tmp")
+        try:
+            with os.fdopen(tmp_fd, "w", encoding="utf-8") as f:
+                json.dump(data, f, ensure_ascii=False, indent=2)
+            os.replace(tmp_path, str(path))
+        except Exception:
+            if os.path.exists(tmp_path):
+                os.unlink(tmp_path)
+            raise

im-client/notifier.py

@@ -0,0 +1,22 @@
+import logging
+
+log = logging.getLogger(__name__)
+
+
+class AgentNotifier:
+    """Agent 通知接口基类。
+
+    当 pending.json 有新消息时，Client 会调用 notify()。
+    每个 Agent 可以继承此类实现自己的通知方式。
+    """
+
+    async def notify(self, count: int, from_contacts: list[str]):
+        raise NotImplementedError
+
+
+class ConsoleNotifier(AgentNotifier):
+    """默认实现：打印到控制台。"""
+
+    async def notify(self, count: int, from_contacts: list[str]):
+        sources = ", ".join(from_contacts)
+        log.info(f"[IM 通知] 你有 {count} 条新消息，来自: {sources}")

im-client/protocol.py

@@ -0,0 +1,23 @@
+from pydantic import BaseModel
+from typing import Optional
+import uuid
+
+
+class IMMessage(BaseModel):
+    msg_id: str = ""
+    sender: str
+    receiver: str
+    content: str
+    msg_type: str = "chat"  # chat | image | video | system
+    sender_chat_id: Optional[str] = None  # 发送方窗口 ID
+    receiver_chat_id: Optional[str] = None  # 接收方窗口 ID（指定则定向，否则广播）
+
+    def model_post_init(self, __context):
+        if not self.msg_id:
+            self.msg_id = uuid.uuid4().hex[:12]
+
+
+class IMResponse(BaseModel):
+    status: str  # "success" | "failed"
+    msg_id: str
+    error: Optional[str] = None

im-client/tools.py

@@ -0,0 +1,227 @@
+"""IM Agent Tools — 供 Agent 在 tool-use loop 中调用的工具函数。
+
+新架构：一个 Agent (contact_id) = 一个 IMClient 实例，该实例管理多个窗口 (chat_id)。
+
+使用方式:
+    1. 调用 setup(contact_id) 初始化 Agent 的 IMClient
+    2. 调用 open_window(contact_id, chat_id) 打开窗口
+    3. 在每轮 loop 中调用 check_notification(contact_id, chat_id) 检查该窗口的新消息
+    4. 有通知时调用 receive_messages(contact_id, chat_id) 读取消息
+    5. 发消息调用 send_message(contact_id, chat_id, receiver, content)
+"""
+
+import asyncio
+import httpx
+
+from client import IMClient
+from notifier import AgentNotifier
+
+# ── 全局状态 ──
+
+_clients: dict[str, IMClient] = {}
+_tasks: dict[str, asyncio.Task] = {}
+_notifications: dict[tuple[str, str], dict] = {}  # (contact_id, chat_id) -> 通知
+
+
+class _ToolNotifier(AgentNotifier):
+    """内部通知器：按 (contact_id, chat_id) 分发通知。"""
+
+    def __init__(self, contact_id: str, chat_id: str):
+        self._key = (contact_id, chat_id)
+
+    async def notify(self, count: int, from_contacts: list[str]):
+        _notifications[self._key] = {"count": count, "from": from_contacts}
+
+
+# ── Tool 1: 初始化 Agent ──
+
+def setup(contact_id: str, server_url: str = "ws://localhost:8000", notify_interval: float = 10.0) -> str:
+    """初始化一个 Agent 的 IMClient（一个实例管理多个窗口）。
+
+    Args:
+        contact_id: Agent 的身份 ID
+        server_url: Server 地址
+        notify_interval: 检查新消息的间隔秒数
+
+    Returns:
+        状态描述
+    """
+    if contact_id in _clients:
+        return f"已连接: {contact_id}"
+
+    client = IMClient(contact_id=contact_id, server_url=server_url, notify_interval=notify_interval)
+    _clients[contact_id] = client
+
+    loop = asyncio.get_event_loop()
+    _tasks[contact_id] = loop.create_task(client.run())
+
+    return f"已启动 IM Client: {contact_id}"
+
+
+def teardown(contact_id: str) -> str:
+    """停止并移除一个 Agent 的 IMClient。"""
+    task = _tasks.pop(contact_id, None)
+    if task:
+        task.cancel()
+    _clients.pop(contact_id, None)
+    # 清理该 contact_id 的所有通知
+    keys_to_remove = [k for k in _notifications if k[0] == contact_id]
+    for k in keys_to_remove:
+        _notifications.pop(k, None)
+    return f"已停止: {contact_id}"
+
+
+# ── Tool 2: 窗口管理 ──
+
+def open_window(contact_id: str, chat_id: str | None = None) -> str:
+    """为某个 Agent 打开一个新窗口。
+
+    Args:
+        contact_id: Agent ID
+        chat_id: 窗口 ID（留空自动生成）
+
+    Returns:
+        窗口的 chat_id
+    """
+    client = _clients.get(contact_id)
+    if client is None:
+        return f"错误: {contact_id} 未初始化"
+
+    actual_chat_id = client.open_window(chat_id=chat_id, notifier=_ToolNotifier(contact_id, chat_id or ""))
+    # 更新 notifier 的 chat_id
+    if chat_id is None:
+        client._notifiers[actual_chat_id] = _ToolNotifier(contact_id, actual_chat_id)
+
+    return actual_chat_id
+
+
+def close_window(contact_id: str, chat_id: str) -> str:
+    """关闭某个窗口。"""
+    client = _clients.get(contact_id)
+    if client is None:
+        return f"错误: {contact_id} 未初始化"
+
+    client.close_window(chat_id)
+    _notifications.pop((contact_id, chat_id), None)
+    return f"已关闭窗口: {chat_id}"
+
+
+def list_windows(contact_id: str) -> list[str]:
+    """列出某个 Agent 的所有窗口。"""
+    client = _clients.get(contact_id)
+    if client is None:
+        return []
+    return client.list_windows()
+
+
+# ── Tool 3: 检查通知 ──
+
+def check_notification(contact_id: str, chat_id: str) -> dict | None:
+    """检查某个窗口是否有新消息通知。
+
+    Returns:
+        有新消息: {"count": 3, "from": ["alice", "bob"]}
+        没有新消息: None
+    """
+    return _notifications.pop((contact_id, chat_id), None)
+
+
+# ── Tool 4: 接收消息 ──
+
+def receive_messages(contact_id: str, chat_id: str) -> list[dict]:
+    """读取某个窗口的待处理消息，读取后自动清空。
+
+    Returns:
+        消息列表，每条格式:
+        {
+            "sender": "alice",
+            "sender_chat_id": "...",
+            "content": "你好",
+            "msg_type": "chat"
+        }
+    """
+    client = _clients.get(contact_id)
+    if client is None:
+        return []
+
+    raw = client.read_pending(chat_id)
+    return [
+        {
+            "sender": m.get("sender", "unknown"),
+            "sender_chat_id": m.get("sender_chat_id"),
+            "content": m.get("content", ""),
+            "msg_type": m.get("msg_type", "chat"),
+        }
+        for m in raw
+    ]
+
+
+# ── Tool 5: 发送消息 ──
+
+def send_message(
+    contact_id: str,
+    chat_id: str,
+    receiver: str,
+    content: str,
+    msg_type: str = "chat",
+    receiver_chat_id: str | None = None,
+) -> str:
+    """从某个窗口发送消息。
+
+    Args:
+        contact_id: 发送方 Agent ID
+        chat_id: 发送方窗口 ID
+        receiver: 接收方 contact_id
+        content: 消息内容
+        msg_type: 消息类型
+        receiver_chat_id: 接收方窗口 ID（不指定则广播）
+
+    Returns:
+        状态描述
+    """
+    client = _clients.get(contact_id)
+    if client is None:
+        return f"错误: {contact_id} 未初始化"
+
+    client.send_message(chat_id, receiver, content, msg_type, receiver_chat_id)
+    target = f"{receiver}:{receiver_chat_id}" if receiver_chat_id else f"{receiver}:*"
+    return f"[{contact_id}:{chat_id}] 已发送给 {target}: {content[:50]}"
+
+
+# ── Tool 6: 查询联系人 ──
+
+def get_contacts(contact_id: str, server_http_url: str = "http://localhost:8000") -> dict:
+    """查询某个 Agent 的联系人列表和在线用户。"""
+    if contact_id not in _clients:
+        return {"error": f"{contact_id} 未初始化"}
+
+    result = {}
+    with httpx.Client() as http:
+        try:
+            r = http.get(f"{server_http_url}/contacts/{contact_id}")
+            result["contacts"] = r.json().get("contacts", [])
+        except Exception as e:
+            result["contacts_error"] = str(e)
+
+        try:
+            r = http.get(f"{server_http_url}/health")
+            result["online"] = r.json().get("online", {})
+        except Exception as e:
+            result["online_error"] = str(e)
+
+    return result
+
+
+# ── Tool 7: 查询聊天历史 ──
+
+def get_chat_history(contact_id: str, chat_id: str, peer_id: str | None = None, limit: int = 20) -> list[dict]:
+    """查询某个窗口的聊天历史。"""
+    client = _clients.get(contact_id)
+    if client is None:
+        return []
+
+    return client.get_chat_history(chat_id, peer_id, limit)
+
+
+
+

test_search.py

@@ -0,0 +1,25 @@
+import asyncio
+import sys
+import io
+
+# 设置 stdout 编码为 UTF-8
+sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
+sys.path.insert(0, '.')
+
+async def test():
+    from agent.tools.builtin.search import search_posts
+
+    print("Testing search_posts with gzh channel...")
+    result = await search_posts(keyword='Midjourney v8', channel='gzh', max_count=20)
+
+    print(f'Title: {result.title}')
+    print(f'Has error: {result.error is not None}')
+
+    if result.error:
+        print(f'Error message: {result.error}')
+    else:
+        print(f'Output length: {len(result.output)} chars')
+        print(f'Has images: {len(result.images) > 0}')
+
+if __name__ == '__main__':
+    asyncio.run(test())