get_unread_sessions
Retrieve all WeChat conversations with unread messages to prioritize responses and manage communication efficiently.
Instructions
获取所有有未读消息的会话
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- Main implementation of the get_unread_sessions tool. Retrieves all sessions with unread messages (new_count > 0) from WeChat and formats them as JSON. Returns session name, time, content, new_count, and ismute for each unread session.
@tool( name="get_unread_sessions", description="获取所有有未读消息的会话" ) def get_unread_sessions() -> str: """ 获取有未读消息的会话 Returns: JSON 格式的会话列表 """ wrapper = _get_wrapper() if not wrapper: return format_result(False, "微信未初始化") try: sessions = wrapper.get_sessions() unread = [s for s in sessions if s.new_count > 0] if not unread: return format_result(True, "无未读消息", {"sessions": []}) formatted = [] for session in unread: formatted.append({ "name": session.name, "time": session.time, "content": session.content, "new_count": session.new_count, "ismute": session.ismute, }) return format_result( True, f"共 {len(unread)} 个会话有未读消息", {"sessions": formatted} ) except Exception as e: logger.error(f"获取未读会话失败: {e}") return format_result(False, f"获取未读会话失败: {e}") - src/wxauto_mcp/wxauto_mcp/api_sessions.py:170-173 (registration)Tool registration using the @tool decorator with name='get_unread_sessions' and description='获取所有有未读消息的会话' (Get all sessions with unread messages).
@tool( name="get_unread_sessions", description="获取所有有未读消息的会话" ) - SessionInfo dataclass schema defining the structure of session data with fields: name, time, content, isnew, new_count, and ismute. Used by get_unread_sessions to represent each unread session.
@dataclass class SessionInfo: """ 会话信息模型 Attributes: name: 聊天名称 time: 最后消息时间 content: 最后一条消息内容 isnew: 是否是新会话 new_count: 未读消息数 ismute: 是否免打扰 """ name: str time: Optional[str] = None content: Optional[str] = None isnew: bool = False new_count: int = 0 ismute: bool = False def to_dict(self) -> dict[str, Any]: """转换为字典""" return { "name": self.name, "time": self.time, "content": self.content, "isnew": self.isnew, "new_count": self.new_count, "ismute": self.ismute } - src/wxauto_mcp/wxauto_mcp/rpc.py:51-80 (registration)The @tool decorator implementation that registers functions as MCP tools. It applies output limiting and registers the tool in TOOL_REGISTRY with its name, function, and description.
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 - format_result helper function used by get_unread_sessions to format the response as JSON with success status, message, and data fields.
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)