wxauto MCP Server

Instructions

Implementation Reference

• src/wxauto_mcp/wxauto_mcp/api_friends.py:175-238 (handler)

  Main handler function for add_friend tool. Decorated with @tool to register it, takes parameters (keywords, addmsg, remark, tags, permission, timeout), validates input, calls wrapper.wx.AddNewFriend(), and returns formatted JSON results.

  @tool(
      name="add_friend",
      description="添加新好友。参数: keywords (搜索关键词，必填), addmsg (验证消息，可选), remark (备注名，可选), tags (标签列表，可选), permission (权限，可选：朋友圈/仅聊天，默认朋友圈), timeout (超时时间，默认5秒)"
  )
  def add_friend(
      keywords: str,
      addmsg: Optional[str] = None,
      remark: Optional[str] = None,
      tags: Optional[list[str]] = None,
      permission: str = "朋友圈",
      timeout: int = 5
  ) -> str:
      """
      添加新好友
  
      Args:
          keywords: 搜索关键词，可以是昵称、微信号、手机号等（必填）
          addmsg: 添加好友时的附加消息（可选）
          remark: 添加好友后的备注（可选）
          tags: 添加好友后的标签列表（可选）
          permission: 添加好友后的权限，可选值：'朋友圈' 或 '仅聊天'（默认'朋友圈'）
          timeout: 超时时间（秒，默认5）
  
      Returns:
          JSON 格式的操作结果
      """
      if not keywords:
          return format_result(False, "搜索关键词不能为空")
  
      wrapper = _get_wrapper()
      if not wrapper:
          return format_result(False, "微信未初始化")
  
      try:
          # 调用 AddNewFriend 方法
          response = wrapper.wx.AddNewFriend(
              keywords=keywords,
              addmsg=addmsg,
              remark=remark,
              tags=tags or [],
              permission=permission,
              timeout=timeout
          )
  
          # 切换回聊天页面
          try:
              wrapper.wx.SwitchToChat()
          except:
              pass  # 如果切换失败，不影响主要功能
  
          # 检查响应
          if hasattr(response, 'success') and response.success:
              return format_result(
                  True,
                  f"已发送好友请求: {keywords}",
                  {"keywords": keywords, "remark": remark, "tags": tags}
              )
          else:
              error_msg = getattr(response, 'message', '添加好友失败')
              return format_result(False, f"添加好友失败: {error_msg}")
  
      except Exception as e:
          logger.error(f"添加好友失败: {e}")
          return format_result(False, f"添加好友失败: {e}")

• src/wxauto_mcp/wxauto_mcp/rpc.py:51-80 (registration)

  The @tool decorator that registers functions as MCP tools. Applies output limiting via limit_output wrapper and registers to TOOL_REGISTRY with 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

• src/wxauto_mcp/wxauto_mcp/utils.py:14-31 (helper)

  Helper function format_result() that formats operation results as JSON strings with success status, message, and optional data dictionary. Used by add_friend 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)

• src/wxauto_mcp/wxauto_mcp/api_friends.py:32-40 (helper)

  Helper function _get_wrapper() that retrieves the WeChatWrapper instance. Used by add_friend to get access to WeChat automation functionality.

  def _get_wrapper():
      """获取 WeChatWrapper 实例"""
      if not WECHAT_AVAILABLE:
          return None
      try:
          return get_wechat()
      except Exception as e:
          logger.error(f"获取 WeChat 实例失败: {e}")
          return None

• src/wxauto_mcp/wxauto_mcp/rpc.py:134-196 (registration)

  ToolRegistry class that manages all registered MCP tools. The register() method stores tool metadata (name, func, description, group, unsafe flag) and maintains groups. Used by the @tool decorator to register add_friend.

  class ToolRegistry:
      """
      工具注册表
  
      管理所有已注册的 MCP 工具，提供扩展分组功能。
      """
  
      def __init__(self) -> None:
          self._tools: dict[str, dict[str, Any]] = {}
          self._groups: dict[str, list[str]] = {}
  
      def register(
          self,
          name: str,
          func: Callable,
          group: str = "default",
          description: Optional[str] = None,
          unsafe: bool = False,
      ) -> None:
          """
          注册工具
  
          Args:
              name: 工具名称
              func: 工具函数
              group: 扩展分组
              description: 工具描述
              unsafe: 是否为不安全操作
          """
          self._tools[name] = {
              "func": func,
              "group": group,
              "description": description,
              "unsafe": unsafe,
          }
  
          if group not in self._groups:
              self._groups[group] = []
          self._groups[group].append(name)
  
          logger.debug(f"已注册工具: {name} (分组: {group}, 不安全: {unsafe})")
  
      def get_tool(self, name: str) -> Optional[dict[str, Any]]:
          """获取指定工具"""
          return self._tools.get(name)
  
      def get_group(self, group: str) -> list[str]:
          """获取指定分组的所有工具名称"""
          return self._groups.get(group, [])
  
      def get_all_tools(self) -> dict[str, dict[str, Any]]:
          """获取所有工具"""
          return self._tools.copy()
  
      def get_all_groups(self) -> dict[str, list[str]]:
          """获取所有分组"""
          return self._groups.copy()
  
      def is_unsafe(self, name: str) -> bool:
          """检查工具是否为不安全操作"""
          tool_info = self._tools.get(name)
          return tool_info.get("unsafe", False) if tool_info else False