get_messages
Retrieve all messages from the current WeChat chat window to analyze conversations or extract information for AI-assisted workflows.
Instructions
获取当前聊天窗口的所有消息
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- The get_messages tool implementation - decorated with @tool for registration and containing the main logic to retrieve all messages from the current chat window, format them, and return as JSON
@tool( name="get_messages", description="获取当前聊天窗口的所有消息" ) def get_messages() -> str: """ 获取当前聊天消息 Returns: JSON 格式的消息列表 """ wrapper = _get_wrapper() if not wrapper: return format_result(False, "微信未初始化") try: messages = wrapper.get_all_messages() if not messages: return format_result(True, "当前聊天无消息", {"messages": []}) # 限制消息数量,避免输出过长 max_messages = 50 display_messages = messages[:max_messages] formatted = [] for msg in display_messages: formatted.append({ "sender": msg.sender, "content": truncate_text(msg.content, 200), "type": msg.msg_type.value if isinstance(msg.msg_type, MessageType) else str(msg.msg_type), "attr": msg.attr.value if isinstance(msg.attr, MessageType) else str(msg.attr), }) result = { "messages": formatted, "total": len(messages), "displayed": len(display_messages), "truncated": len(messages) > max_messages } return format_result(True, f"获取成功,共 {len(messages)} 条消息", result) except Exception as e: logger.error(f"获取消息失败: {e}") return format_result(False, f"获取消息失败: {e}") - src/wxauto_mcp/wxauto_mcp/rpc.py:51-81 (registration)The @tool decorator that registers functions as MCP tools, applying output limiting and registering to TOOL_REGISTRY for automatic discovery
def tool(name: Optional[str] = None, description: Optional[str] = None) -> Callable: """ 工具装饰器 注册函数为 MCP 工具,自动应用输出限制。 Args: name: 工具名称,默认使用函数名 description: 工具描述 Returns: 装饰后的函数 """ def decorator(func: Callable) -> Callable: # 应用输出限制 limited_func = limit_output(func) # 注册到工具注册表 tool_name = name or func.__name__ TOOL_REGISTRY.register( name=tool_name, func=limited_func, description=description or func.__doc__ or "", ) # 返回原函数,不修改其行为 return func return decorator - MessageInfo dataclass defining the message structure (content, sender, msg_type, attr, timestamp, chat_name) used by get_messages
class MessageInfo: """ 消息信息模型 Attributes: content: 消息内容 sender: 发送者 msg_type: 消息类型 attr: 消息属性 timestamp: 时间戳 chat_name: 所属聊天名称 """ content: str sender: str msg_type: MessageType attr: MessageAttribute timestamp: Optional[float] = None chat_name: Optional[str] = None def to_dict(self) -> dict[str, Any]: """转换为字典""" return { "content": self.content, "sender": self.sender, "type": self.msg_type.value if isinstance(self.msg_type, MessageType) else self.msg_type, "attr": self.attr.value if isinstance(self.attr, MessageAttribute) else self.attr, "timestamp": self.timestamp, "chat_name": self.chat_name } - MessageType and MessageAttribute enums defining the types of messages (TEXT, IMAGE, VIDEO, FILE, VOICE) and attributes (SYSTEM, SELF, FRIEND) used in message formatting
class MessageType(str, Enum): """消息类型枚举""" TEXT = "text" IMAGE = "image" VIDEO = "video" FILE = "file" VOICE = "voice" class MessageAttribute(str, Enum): """消息属性枚举""" SYSTEM = "system" SELF = "self" FRIEND = "friend" - format_result helper function that formats operation results as JSON strings, used by get_messages to return standardized responses
def format_result(success: bool, message: str, data: Optional[dict[str, Any]] = None) -> str: """ 格式化操作结果为 JSON 字符串 Args: success: 操作是否成功 message: 结果消息 data: 附加数据 Returns: JSON 格式的结果字符串 """ result = { "success": success, "message": message, "data": data or {}, } return json.dumps(result, ensure_ascii=False, indent=2)