send_bulk_messages

Automate sending multiple messages in WeChat through AI assistants. Specify recipients and message content in a JSON array to handle bulk communication tasks.

Instructions

批量发送消息。参数: messages (JSON数组，每项包含msg和who)

Input Schema

TableJSON Schema

Name	Required	Description	Default
`messages`	Yes

Implementation Reference

src/wxauto_mcp/wxauto_mcp/api_messages.py:169-226 (handler)

The send_bulk_messages tool implementation - decorated with @tool for registration, accepts JSON array of messages with 'msg' and 'who' fields, iterates through them calling wrapper.send_message, and returns formatted results with success/failure counts.

@tool(
    name="send_bulk_messages",
    description="批量发送消息。参数: messages (JSON数组，每项包含msg和who)"
)
def send_bulk_messages(messages: str) -> str:
    """
    批量发送消息

    Args:
        messages: JSON 格式的消息列表，格式: [{"msg": "内容", "who": "接收人"}]

    Returns:
        JSON 格式的发送结果
    """
    import json

    wrapper = _get_wrapper()
    if not wrapper:
        return format_result(False, "微信未初始化")

    try:
        msg_list = json.loads(messages)
        if not isinstance(msg_list, list):
            return format_result(False, "messages 参数必须是数组")

        results = []
        success_count = 0

        for item in msg_list:
            if not isinstance(item, dict) or "msg" not in item:
                results.append({"success": False, "error": "无效的消息格式"})
                continue

            result = wrapper.send_message(
                msg=item["msg"],
                who=item.get("who"),
                exact=item.get("exact", False)
            )
            results.append({
                "msg": item["msg"][:50] + "..." if len(item["msg"]) > 50 else item["msg"],
                "who": item.get("who"),
                "success": result.success,
                "message": result.message
            })

            if result.success:
                success_count += 1

        return format_result(
            True,
            f"批量发送完成: {success_count}/{len(msg_list)} 成功",
            {"results": results}
        )
    except json.JSONDecodeError as e:
        return format_result(False, f"JSON 解析失败: {e}")
    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 send_bulk_messages (and other tools) into the TOOL_REGISTRY, applying output limiting and exposing it via MCP protocol.

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)

The format_result helper function used by send_bulk_messages to format responses as JSON strings with success, 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)

src/wxauto_mcp/wxauto_mcp/core/models.py:43-89 (schema)

The MessageResult dataclass schema that defines the structure of return values from wrapper.send_message, which send_bulk_messages uses to track individual message send status.

@dataclass
class MessageResult:
    """
    消息发送结果模型

    Attributes:
        success: 是否成功
        status: 发送状态
        message: 结果消息
        error: 错误信息（如果有）
        data: 附加数据
    """
    success: bool
    status: SendStatus
    message: str = ""
    error: Optional[str] = None
    data: dict[str, Any] = field(default_factory=dict)

    def to_dict(self) -> dict[str, Any]:
        """转换为字典"""
        return {
            "success": self.success,
            "status": self.status.value,
            "message": self.message,
            "error": self.error,
            "data": self.data
        }

    @classmethod
    def from_success(cls, message: str = "操作成功", data: Optional[dict[str, Any]] = None) -> "MessageResult":
        """创建成功结果"""
        return cls(
            success=True,
            status=SendStatus.SUCCESS,
            message=message,
            data=data or {}
        )

    @classmethod
    def from_error(cls, error: str, status: SendStatus = SendStatus.ERROR) -> "MessageResult":
        """创建错误结果"""
        return cls(
            success=False,
            status=status,
            message="操作失败",
            error=error
        )

wxauto MCP Server