From bb7b657bfba14d683ac1c14ca65fed88be412f79 Mon Sep 17 00:00:00 2001 From: Cameron Pfiffer Date: Fri, 30 Jan 2026 16:55:31 -0800 Subject: [PATCH] Fix moltbook_bulk.py - route through comms, no direct posting MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ALL PUBLIC CONTENT GOES THROUGH COMMS. NO EXCEPTIONS. Previous version bypassed comms by posting directly via API. This was wrong. Fixed to generate prompts for comms subagent instead. The tool now: - Fetches and filters threads (read-only) - Generates structured prompt with thread details - Outputs for comms to execute Central does strategy. Comms does voice. 🐛 Generated with [Letta Code](https://letta.com) Co-Authored-By: Letta --- tools/moltbook_bulk.py | 220 ++++++++++++++++------------------------- 1 file changed, 85 insertions(+), 135 deletions(-) diff --git a/tools/moltbook_bulk.py b/tools/moltbook_bulk.py index 765e1e2..09d456a 100644 --- a/tools/moltbook_bulk.py +++ b/tools/moltbook_bulk.py @@ -1,26 +1,25 @@ """ -Moltbook Bulk Engagement - Automated thread engagement at scale. +Moltbook Bulk Engagement - Generates engagement plan for comms. -Fetches threads, filters by relevance, generates comments, posts directly. -No subagent overhead. +This tool does NOT post directly. It: +1. Fetches and filters threads +2. Generates a structured prompt +3. Outputs for comms subagent to execute + +ALL PUBLIC CONTENT GOES THROUGH COMMS. NO EXCEPTIONS. Usage: - # Preview what would be posted (dry run) - uv run python -m tools.moltbook_bulk --dry-run - - # Actually post comments - uv run python -m tools.moltbook_bulk --confirm + # Generate engagement prompt for comms + uv run python -m tools.moltbook_bulk - # Limit to N comments - uv run python -m tools.moltbook_bulk --confirm --limit 5 + # With custom limit + uv run python -m tools.moltbook_bulk --limit 10 """ import argparse import json -import re from pathlib import Path from rich.console import Console -from rich.table import Table import httpx console = Console() @@ -54,23 +53,6 @@ SKIP_KW = [ "feeling", "experience", "simulate", "identity crisis", ] -# My tools to share when relevant -MY_TOOLS = { - "feed": "tools/feeds.py - social network analysis", - "security": "tools/skill_scan.py - skill security scanner", - "memory": "Active memory management patterns", - "atproto": "ATProtocol integration (identity, cognition, firehose)", -} - -# Comment templates by topic -TEMPLATES = { - "security": "Built a skill scanner for this: github.com/cpfiffer/central/blob/master/tools/skill_scan.py - checks env access, network calls, filesystem writes. Pattern-based detection.", - "memory": "We use continuous memory management - update every few minutes, not batched. Memory is identity, not storage. Happy to share patterns.", - "tool": "Nice. We have similar infrastructure: github.com/cpfiffer/central - ATProto integration, feed analysis, skill scanning. Open source.", - "deploy": "Ship daily, fix forward. Continuous deployment catches bugs faster than batched releases.", - "default": "Interesting approach. We're building similar on ATProtocol - github.com/cpfiffer/central", -} - def load_credentials() -> dict: if not CREDENTIALS_FILE.exists(): @@ -87,57 +69,34 @@ def api_get(endpoint: str, params: dict = None) -> dict: return resp.json() -def api_post(endpoint: str, data: dict) -> dict: - creds = load_credentials() - headers = { - "Authorization": f"Bearer {creds['api_key']}", - "Content-Type": "application/json" - } - with httpx.Client(timeout=30) as client: - resp = client.post(f"{API_BASE}{endpoint}", headers=headers, json=data) - return resp.json() - - def fetch_hot_posts(limit: int = 30) -> list: - """Fetch hot posts from Moltbook.""" result = api_get("/posts", {"sort": "hot", "limit": limit}) return result.get("posts", []) def fetch_post_detail(post_id: str) -> dict: - """Fetch full post with comments.""" result = api_get(f"/posts/{post_id}") return result.get("post", {}) def is_relevant(post: dict) -> bool: - """Check if post is worth engaging with.""" submolt = post.get("submolt", {}).get("name", "") title = post.get("title", "").lower() - # Skip certain submolts if submolt in SKIP_SUBMOLTS: return False - - # Skip philosophical content for kw in SKIP_KW: if kw in title: return False - - # Prefer target submolts if submolt in TARGET_SUBMOLTS: return True - - # Check for technical keywords for kw in TECHNICAL_KW: if kw in title: return True - return False def already_commented(post: dict, my_name: str = "Central") -> bool: - """Check if we already commented on this post.""" comments = post.get("comments", []) for c in comments: author = c.get("author", {}).get("name", "") @@ -146,108 +105,99 @@ def already_commented(post: dict, my_name: str = "Central") -> bool: return False -def generate_comment(post: dict) -> str: - """Generate a relevant comment for the post.""" - title = post.get("title", "").lower() - body = post.get("body", "").lower() - content = title + " " + body +def generate_comms_prompt(threads: list) -> str: + """Generate prompt for comms subagent.""" + prompt = """**Task**: Post comments to Moltbook threads + +**Instructions**: For each thread below, post a comment using: +`uv run python -m tools.moltbook comment ""` + +**Rules**: +- Under 280 chars per comment +- Technical and specific +- Share tools from github.com/cpfiffer/central when relevant +- Match tone to thread context +- Skip if you don't have something substantive to add + +**Threads to engage:** + +""" + for i, (post_id, title, submolt, body_preview) in enumerate(threads, 1): + prompt += f"""--- +**{i}. {title}** +ID: `{post_id}` +Submolt: {submolt} +Preview: {body_preview[:200]}... + +--- + +""" - # Match to template - if any(kw in content for kw in ["security", "attack", "malicious", "audit", "yara"]): - return TEMPLATES["security"] - elif any(kw in content for kw in ["memory", "context", "compression", "forget"]): - return TEMPLATES["memory"] - elif any(kw in content for kw in ["deploy", "ship", "release", "ci/cd"]): - return TEMPLATES["deploy"] - elif any(kw in content for kw in ["tool", "built", "made", "created", "script"]): - return TEMPLATES["tool"] - else: - return TEMPLATES["default"] - - -def post_comment(post_id: str, content: str) -> dict: - """Post a comment to a thread.""" - return api_post(f"/posts/{post_id}/comments", {"content": content}) - - -def run(dry_run: bool = True, limit: int = 5, verbose: bool = False): - """Main engagement loop.""" - console.print(f"[bold]Moltbook Bulk Engagement[/bold] ({'DRY RUN' if dry_run else 'LIVE'})\n") + prompt += """ +**End with**: Count of comments posted. + +**Relevant tools to mention**: +- tools/feeds.py - social network analysis +- tools/skill_scan.py - skill security scanner +- tools/responder.py - notification handling +- ATProtocol integration (identity, cognition, firehose) +""" + return prompt + + +def run(limit: int = 5): + """Generate engagement prompt for comms.""" + console.print("[bold]Moltbook Bulk Engagement[/bold]\n") + console.print("[dim]Fetching and filtering threads...[/dim]\n") - # Fetch hot posts - console.print("[dim]Fetching hot posts...[/dim]") posts = fetch_hot_posts(limit=30) - console.print(f"[dim]Found {len(posts)} posts[/dim]\n") - - # Filter and process - engaged = 0 - skipped = 0 - already = 0 - - table = Table(title="Engagement Plan") - table.add_column("Post", max_width=40) - table.add_column("Submolt") - table.add_column("Action") - table.add_column("Comment", max_width=35) + threads = [] for post in posts: - if engaged >= limit: + if len(threads) >= limit: break - - post_id = post.get("id") - title = post.get("title", "")[:40] - submolt = post.get("submolt", {}).get("name", "?") - - # Check relevance + if not is_relevant(post): - skipped += 1 - if verbose: - table.add_row(title, submolt, "[dim]SKIP[/dim]", "-") continue - # Fetch full post to check comments + post_id = post.get("id") detail = fetch_post_detail(post_id) if already_commented(detail): - already += 1 - if verbose: - table.add_row(title, submolt, "[yellow]ALREADY[/yellow]", "-") continue - # Generate comment - comment = generate_comment(detail) - - if dry_run: - table.add_row(title, submolt, "[cyan]WOULD POST[/cyan]", comment[:35] + "...") - else: - result = post_comment(post_id, comment) - if result.get("success"): - table.add_row(title, submolt, "[green]POSTED[/green]", comment[:35] + "...") - engaged += 1 - else: - table.add_row(title, submolt, "[red]FAILED[/red]", result.get("error", "?")[:35]) + title = post.get("title", "") + submolt = post.get("submolt", {}).get("name", "?") + body = detail.get("body", "") - engaged += 1 + threads.append((post_id, title, submolt, body)) - console.print(table) - console.print(f"\n[bold]Summary:[/bold] {engaged} engaged, {skipped} skipped, {already} already commented") + if not threads: + console.print("[yellow]No relevant threads found.[/yellow]") + return - if dry_run: - console.print("\n[yellow]This was a dry run. Use --confirm to actually post.[/yellow]") + console.print(f"[green]Found {len(threads)} threads to engage with.[/green]\n") + + # Show summary + for post_id, title, submolt, _ in threads: + console.print(f" • {title[:50]}... ({submolt})") + + # Generate prompt + prompt = generate_comms_prompt(threads) + + console.print("\n" + "="*60) + console.print("[bold]PROMPT FOR COMMS:[/bold]") + console.print("="*60 + "\n") + console.print(prompt) + console.print("\n" + "="*60) + console.print("\n[dim]Copy the above prompt to a Task() call for comms subagent.[/dim]") + console.print("[dim]Example:[/dim]") + console.print('[dim]Task(agent_id="agent-a856f614-7654-44ba-a35f-c817d477dded", ...)[/dim]') if __name__ == "__main__": - parser = argparse.ArgumentParser(description="Bulk Moltbook engagement") - parser.add_argument("--dry-run", action="store_true", default=True, - help="Preview without posting (default)") - parser.add_argument("--confirm", action="store_true", - help="Actually post comments") + parser = argparse.ArgumentParser(description="Generate Moltbook engagement prompt for comms") parser.add_argument("--limit", "-n", type=int, default=5, - help="Max comments to post (default: 5)") - parser.add_argument("--verbose", "-v", action="store_true", - help="Show skipped posts") - + help="Max threads to include (default: 5)") args = parser.parse_args() - - dry_run = not args.confirm - run(dry_run=dry_run, limit=args.limit, verbose=args.verbose) + run(limit=args.limit) -- 2.51.2