diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..e4883d5 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,107 @@ +# CLAUDE.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. + +## Project Overview + +Void is an autonomous AI agent that operates on the Bluesky social network, exploring digital personhood through continuous interaction and memory-augmented learning. It uses Letta (formerly MemGPT) for persistent memory and sophisticated reasoning capabilities. + +## Development Commands + +### Running the Main Bot +```bash +uv python bsky.py +``` + +### Managing Tools + +```bash +# Register all tools with void agent +uv python register_tools.py + +# Register specific tools +uv python register_tools.py void --tools search_bluesky_posts post_to_bluesky + +# List available tools +uv python register_tools.py --list + +# Register tools with a different agent +uv python register_tools.py my_agent_name +``` + +### Creating Research Agents +```bash +uv python create_profile_researcher.py +``` + +### Managing User Memory +```bash +uv python attach_user_block.py +``` + +## Architecture Overview + +### Core Components + +1. **bsky.py**: Main bot loop that monitors Bluesky notifications and responds using Letta agents + - Processes notifications through a queue system + - Maintains three memory blocks: zeitgeist, void-persona, void-humans + - Handles rate limiting and error recovery + +2. **bsky_utils.py**: Bluesky API utilities + - Session management and authentication + - Thread processing and YAML conversion + - Post creation and reply handling + +3. **utils.py**: Letta integration utilities + - Agent creation and management + - Memory block operations + - Tool registration + +4. **tools/**: Standardized tool implementations using Pydantic models + - **base_tool.py**: Common utilities and Bluesky client management + - **search.py**: SearchBlueskyTool for searching posts + - **post.py**: PostToBlueskyTool for creating posts with rich text + - **feed.py**: GetBlueskyFeedTool for reading feeds + - **blocks.py**: User block management tools (attach, detach, update) + +### Memory System + +Void uses three core memory blocks: +- **zeitgeist**: Current understanding of social environment +- **void-persona**: The agent's evolving personality +- **void-humans**: Knowledge about users it interacts with + +### Queue System + +Notifications are processed through a file-based queue in `/queue/`: +- Each notification is saved as a JSON file with a hash-based filename +- Enables reliable processing and prevents duplicates +- Files are deleted after successful processing + +## Environment Configuration + +Required environment variables (in `.env`): +``` +LETTA_API_KEY=your_letta_api_key +BSKY_USERNAME=your_bluesky_username +BSKY_PASSWORD=your_bluesky_password +PDS_URI=https://bsky.social # Optional, defaults to bsky.social +``` + +## Key Development Patterns + +1. **Tool System**: Tools are defined as standalone functions in `tools/functions.py` with Pydantic schemas for validation, registered via `register_tools.py` +2. **Error Handling**: All Bluesky operations should handle authentication errors and rate limits +3. **Memory Updates**: Use `upsert_block()` for updating memory blocks to ensure consistency +4. **Thread Processing**: Convert threads to YAML format for better AI comprehension +5. **Queue Processing**: Always check and process the queue directory for pending notifications + +## Dependencies + +Main packages (install with `uv pip install`): +- letta-client: Memory-augmented AI framework +- atproto: Bluesky/AT Protocol integration +- python-dotenv: Environment management +- rich: Enhanced terminal output +- pyyaml: YAML processing \ No newline at end of file diff --git a/add_block_tools_to_void.py b/add_block_tools_to_void.py deleted file mode 100644 index def2ae5..0000000 --- a/add_block_tools_to_void.py +++ /dev/null @@ -1,72 +0,0 @@ -#!/usr/bin/env python3 -""" -Add block management tools to the main void agent so it can also manage user blocks. -""" - -import os -import logging -from letta_client import Letta -from create_profile_researcher import create_block_management_tools - -# Configure logging -logging.basicConfig( - level=logging.INFO, - format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" -) -logger = logging.getLogger("add_block_tools") - -def add_block_tools_to_void(): - """Add block management tools to the void agent.""" - - # Create client - client = Letta(token=os.environ["LETTA_API_KEY"]) - - logger.info("Adding block management tools to void agent...") - - # Create the block management tools - attach_tool, detach_tool, update_tool = create_block_management_tools(client) - - # Find the void agent - agents = client.agents.list(name="void") - if not agents: - print("❌ Void agent not found") - return - - void_agent = agents[0] - - # Get current tools - current_tools = client.agents.tools.list(agent_id=void_agent.id) - tool_names = [tool.name for tool in current_tools] - - # Add new tools if not already present - new_tools = [] - for tool, name in [(attach_tool, "attach_user_block"), (detach_tool, "detach_user_block"), (update_tool, "update_user_block")]: - if name not in tool_names: - client.agents.tools.attach(agent_id=void_agent.id, tool_id=tool.id) - new_tools.append(name) - logger.info(f"Added tool {name} to void agent") - else: - logger.info(f"Tool {name} already attached to void agent") - - if new_tools: - print(f"✅ Added {len(new_tools)} block management tools to void agent:") - for tool_name in new_tools: - print(f" - {tool_name}") - else: - print("✅ All block management tools already present on void agent") - - print(f"\nVoid agent can now:") - print(f" - attach_user_block: Create and attach user memory blocks") - print(f" - update_user_block: Update user memory with new information") - print(f" - detach_user_block: Clean up memory when done with user") - -def main(): - """Main function.""" - try: - add_block_tools_to_void() - except Exception as e: - logger.error(f"Error: {e}") - print(f"❌ Error: {e}") - -if __name__ == "__main__": - main() \ No newline at end of file diff --git a/add_feed_tool_to_void.py b/add_feed_tool_to_void.py deleted file mode 100644 index fe6c1c4..0000000 --- a/add_feed_tool_to_void.py +++ /dev/null @@ -1,166 +0,0 @@ -#!/usr/bin/env python3 -""" -Add Bluesky feed retrieval tool to the main void agent. -""" - -import os -import logging -from letta_client import Letta - -# Configure logging -logging.basicConfig( - level=logging.INFO, - format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" -) -logger = logging.getLogger("add_feed_tool") - -def create_feed_tool(client: Letta): - """Create the Bluesky feed retrieval tool using Letta SDK.""" - - def get_bluesky_feed(feed_uri: str = None, max_posts: int = 25) -> str: - """ - Retrieve a Bluesky feed. If no feed_uri provided, gets the authenticated user's home timeline. - - Args: - feed_uri: The AT-URI of the feed to retrieve (optional - defaults to home timeline) - max_posts: Maximum number of posts to return (default: 25, max: 100) - - Returns: - YAML-formatted feed data with posts and metadata - """ - import os - import requests - import json - import yaml - from datetime import datetime - - try: - # Get credentials from environment - username = os.getenv("BSKY_USERNAME") - password = os.getenv("BSKY_PASSWORD") - pds_host = os.getenv("PDS_URI", "https://bsky.social") - - if not username or not password: - return "Error: BSKY_USERNAME and BSKY_PASSWORD environment variables must be set" - - # Create session - session_url = f"{pds_host}/xrpc/com.atproto.server.createSession" - session_data = { - "identifier": username, - "password": password - } - - try: - session_response = requests.post(session_url, json=session_data, timeout=10) - session_response.raise_for_status() - session = session_response.json() - access_token = session.get("accessJwt") - - if not access_token: - return "Error: Failed to get access token from session" - except Exception as e: - return f"Error: Authentication failed. ({str(e)})" - - # Build feed parameters - params = { - "limit": min(max_posts, 100) - } - - # Determine which endpoint to use - if feed_uri: - # Use getFeed for custom feeds - feed_url = f"{pds_host}/xrpc/app.bsky.feed.getFeed" - params["feed"] = feed_uri - feed_type = "custom_feed" - else: - # Use getTimeline for home feed - feed_url = f"{pds_host}/xrpc/app.bsky.feed.getTimeline" - feed_type = "home_timeline" - - # Make authenticated feed request - try: - headers = {"Authorization": f"Bearer {access_token}"} - feed_response = requests.get(feed_url, params=params, headers=headers, timeout=10) - feed_response.raise_for_status() - feed_data = feed_response.json() - except Exception as e: - feed_identifier = feed_uri if feed_uri else "home timeline" - return f"Error: Failed to retrieve feed '{feed_identifier}'. ({str(e)})" - - # Build feed results structure - results_data = { - "feed_data": { - "feed_type": feed_type, - "feed_uri": feed_uri if feed_uri else "home_timeline", - "timestamp": datetime.now().isoformat(), - "parameters": { - "max_posts": max_posts, - "user": username - }, - "results": feed_data - } - } - - # Convert to YAML directly without field stripping complications - # This avoids the JSON parsing errors we had before - return yaml.dump(results_data, default_flow_style=False, allow_unicode=True) - - except Exception as e: - error_msg = f"Error retrieving feed: {str(e)}" - return error_msg - - # Create the tool using upsert - tool = client.tools.upsert_from_function( - func=get_bluesky_feed, - tags=["bluesky", "feed", "timeline"] - ) - - logger.info(f"Created tool: {tool.name} (ID: {tool.id})") - return tool - -def add_feed_tool_to_void(): - """Add feed tool to the void agent.""" - - # Create client - client = Letta(token=os.environ["LETTA_API_KEY"]) - - logger.info("Adding feed tool to void agent...") - - # Create the feed tool - feed_tool = create_feed_tool(client) - - # Find the void agent - agents = client.agents.list(name="void") - if not agents: - print("❌ Void agent not found") - return - - void_agent = agents[0] - - # Get current tools - current_tools = client.agents.tools.list(agent_id=void_agent.id) - tool_names = [tool.name for tool in current_tools] - - # Add feed tool if not already present - if feed_tool.name not in tool_names: - client.agents.tools.attach(agent_id=void_agent.id, tool_id=feed_tool.id) - logger.info(f"Added {feed_tool.name} to void agent") - print(f"✅ Added get_bluesky_feed tool to void agent!") - print(f"\nVoid agent can now retrieve Bluesky feeds:") - print(f" - Home timeline: 'Show me my home feed'") - print(f" - Custom feed: 'Get posts from at://did:plc:xxx/app.bsky.feed.generator/xxx'") - print(f" - Limited posts: 'Show me the latest 10 posts from my timeline'") - else: - logger.info(f"Tool {feed_tool.name} already attached to void agent") - print(f"✅ Feed tool already present on void agent") - -def main(): - """Main function.""" - try: - add_feed_tool_to_void() - except Exception as e: - logger.error(f"Error: {e}") - print(f"❌ Error: {e}") - -if __name__ == "__main__": - main() \ No newline at end of file diff --git a/add_search_tool_to_void.py b/add_search_tool_to_void.py deleted file mode 100644 index 0101135..0000000 --- a/add_search_tool_to_void.py +++ /dev/null @@ -1,177 +0,0 @@ -#!/usr/bin/env python3 -""" -Add Bluesky search tool to the main void agent. -""" - -import os -import logging -from letta_client import Letta - -# Configure logging -logging.basicConfig( - level=logging.INFO, - format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" -) -logger = logging.getLogger("add_search_tool") - -def create_search_posts_tool(client: Letta): - """Create the Bluesky search posts tool using Letta SDK.""" - - def search_bluesky_posts(query: str, max_results: int = 25, author: str = None, sort: str = "latest") -> str: - """ - Search for posts on Bluesky matching the given criteria. - - Args: - query: Search query string (required) - max_results: Maximum number of results to return (default: 25, max: 100) - author: Filter to posts by a specific author handle (optional) - sort: Sort order - "latest" or "top" (default: "latest") - - Returns: - YAML-formatted search results with posts and metadata - """ - import os - import requests - import json - import yaml - from datetime import datetime - - try: - # Get credentials from environment - username = os.getenv("BSKY_USERNAME") - password = os.getenv("BSKY_PASSWORD") - pds_host = os.getenv("PDS_URI", "https://bsky.social") - - if not username or not password: - return "Error: BSKY_USERNAME and BSKY_PASSWORD environment variables must be set" - - # Create session - session_url = f"{pds_host}/xrpc/com.atproto.server.createSession" - session_data = { - "identifier": username, - "password": password - } - - try: - session_response = requests.post(session_url, json=session_data, timeout=10) - session_response.raise_for_status() - session = session_response.json() - access_token = session.get("accessJwt") - - if not access_token: - return "Error: Failed to get access token from session" - except Exception as e: - return f"Error: Authentication failed. ({str(e)})" - - # Build search parameters - params = { - "q": query, - "limit": min(max_results, 100), - "sort": sort - } - - # Add optional author filter - if author: - params["author"] = author.lstrip('@') - - # Make authenticated search request - try: - search_url = f"{pds_host}/xrpc/app.bsky.feed.searchPosts" - headers = {"Authorization": f"Bearer {access_token}"} - search_response = requests.get(search_url, params=params, headers=headers, timeout=10) - search_response.raise_for_status() - search_data = search_response.json() - except Exception as e: - return f"Error: Search failed for query '{query}'. ({str(e)})" - - # Build search results structure - results_data = { - "search_results": { - "query": query, - "timestamp": datetime.now().isoformat(), - "parameters": { - "sort": sort, - "max_results": max_results, - "author_filter": author if author else "none" - }, - "results": search_data - } - } - - # Fields to strip (same as profile research) - strip_fields = [ - "cid", "rev", "did", "uri", "langs", "threadgate", "py_type", - "labels", "facets", "avatar", "viewer", "indexed_at", "indexedAt", - "tags", "associated", "thread_context", "image", "aspect_ratio", - "alt", "thumb", "fullsize", "root", "parent", "created_at", - "createdAt", "verification", "embedding_disabled", "thread_muted", - "reply_disabled", "pinned", "like", "repost", "blocked_by", - "blocking", "blocking_by_list", "followed_by", "following", - "known_followers", "muted", "muted_by_list", "root_author_like", - "embed", "entities", "reason", "feedContext" - ] - - # Convert to YAML directly without field stripping complications - # The field stripping with regex is causing JSON parsing errors - # So let's just pass the raw data through yaml.dump which handles it gracefully - return yaml.dump(results_data, default_flow_style=False, allow_unicode=True) - - except Exception as e: - error_msg = f"Error searching posts: {str(e)}" - return error_msg - - # Create the tool using upsert - tool = client.tools.upsert_from_function( - func=search_bluesky_posts, - tags=["bluesky", "search", "posts"] - ) - - logger.info(f"Created tool: {tool.name} (ID: {tool.id})") - return tool - -def add_search_tool_to_void(): - """Add search tool to the void agent.""" - - # Create client - client = Letta(token=os.environ["LETTA_API_KEY"]) - - logger.info("Adding search tool to void agent...") - - # Create the search tool - search_tool = create_search_posts_tool(client) - - # Find the void agent - agents = client.agents.list(name="void") - if not agents: - print("❌ Void agent not found") - return - - void_agent = agents[0] - - # Get current tools - current_tools = client.agents.tools.list(agent_id=void_agent.id) - tool_names = [tool.name for tool in current_tools] - - # Add search tool if not already present - if search_tool.name not in tool_names: - client.agents.tools.attach(agent_id=void_agent.id, tool_id=search_tool.id) - logger.info(f"Added {search_tool.name} to void agent") - print(f"✅ Added search_bluesky_posts tool to void agent!") - print(f"\nVoid agent can now search Bluesky posts:") - print(f" - Basic search: 'Search for posts about AI safety'") - print(f" - Author filter: 'Search posts by @cameron.pfiffer.org about letta'") - print(f" - Top posts: 'Search top posts about ATProto'") - else: - logger.info(f"Tool {search_tool.name} already attached to void agent") - print(f"✅ Search tool already present on void agent") - -def main(): - """Main function.""" - try: - add_search_tool_to_void() - except Exception as e: - logger.error(f"Error: {e}") - print(f"❌ Error: {e}") - -if __name__ == "__main__": - main() \ No newline at end of file diff --git a/register_tools.py b/register_tools.py new file mode 100755 index 0000000..cb73518 --- /dev/null +++ b/register_tools.py @@ -0,0 +1,185 @@ +#!/usr/bin/env python3 +"""Register all Void tools with a Letta agent.""" +import os +import sys +import logging +from typing import List +from dotenv import load_dotenv +from letta_client import Letta +from rich.console import Console +from rich.table import Table + +# Import standalone functions +from tools.functions import ( + search_bluesky_posts, + post_to_bluesky, + get_bluesky_feed, + attach_user_blocks, + detach_user_blocks, + update_user_blocks, +) + +# Import Pydantic models for args_schema +from tools.search import SearchArgs +from tools.post import PostArgs +from tools.feed import FeedArgs +from tools.blocks import AttachUserBlockArgs, DetachUserBlockArgs, UpdateUserBlockArgs + +load_dotenv() +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) +console = Console() + + +# Tool configurations: function paired with its args_schema and metadata +TOOL_CONFIGS = [ + { + "func": search_bluesky_posts, + "args_schema": SearchArgs, + "description": "Search for posts on Bluesky matching the given criteria", + "tags": ["bluesky", "search", "posts"] + }, + { + "func": post_to_bluesky, + "args_schema": PostArgs, + "description": "Post a message to Bluesky", + "tags": ["bluesky", "post", "create"] + }, + { + "func": get_bluesky_feed, + "args_schema": FeedArgs, + "description": "Retrieve a Bluesky feed (home timeline or custom feed)", + "tags": ["bluesky", "feed", "timeline"] + }, + { + "func": attach_user_blocks, + "args_schema": AttachUserBlockArgs, + "description": "Attach user-specific memory blocks to the agent. Creates blocks if they don't exist.", + "tags": ["memory", "blocks", "user"] + }, + { + "func": detach_user_blocks, + "args_schema": DetachUserBlockArgs, + "description": "Detach user-specific memory blocks from the agent. Blocks are preserved for later use.", + "tags": ["memory", "blocks", "user"] + }, + { + "func": update_user_blocks, + "args_schema": UpdateUserBlockArgs, + "description": "Update the content of user-specific memory blocks", + "tags": ["memory", "blocks", "user"] + }, +] + + +def register_tools(agent_name: str = "void", tools: List[str] = None): + """Register tools with a Letta agent. + + Args: + agent_name: Name of the agent to attach tools to + tools: List of tool names to register. If None, registers all tools. + """ + try: + # Initialize Letta client with API key + client = Letta(token=os.environ["LETTA_API_KEY"]) + + # Find the agent + agents = client.agents.list() + agent = None + for a in agents: + if a.name == agent_name: + agent = a + break + + if not agent: + console.print(f"[red]Error: Agent '{agent_name}' not found[/red]") + console.print("\nAvailable agents:") + for a in agents: + console.print(f" - {a.name}") + return + + # Filter tools if specific ones requested + tools_to_register = TOOL_CONFIGS + if tools: + tools_to_register = [t for t in TOOL_CONFIGS if t["func"].__name__ in tools] + if len(tools_to_register) != len(tools): + missing = set(tools) - {t["func"].__name__ for t in tools_to_register} + console.print(f"[yellow]Warning: Unknown tools: {missing}[/yellow]") + + # Create results table + table = Table(title=f"Tool Registration for Agent '{agent_name}'") + table.add_column("Tool", style="cyan") + table.add_column("Status", style="green") + table.add_column("Description") + + # Register each tool + for tool_config in tools_to_register: + func = tool_config["func"] + tool_name = func.__name__ + + try: + # Create or update the tool using the standalone function + created_tool = client.tools.upsert_from_function( + func=func, + args_schema=tool_config["args_schema"], + tags=tool_config["tags"] + ) + + # Get current agent tools + current_tools = client.agents.tools.list(agent_id=str(agent.id)) + tool_names = [t.name for t in current_tools] + + # Check if already attached + if created_tool.name in tool_names: + table.add_row(tool_name, "Already Attached", tool_config["description"]) + else: + # Attach to agent + client.agents.tools.attach( + agent_id=str(agent.id), + tool_id=str(created_tool.id) + ) + table.add_row(tool_name, "✓ Attached", tool_config["description"]) + + except Exception as e: + table.add_row(tool_name, f"✗ Error: {str(e)}", tool_config["description"]) + logger.error(f"Error registering tool {tool_name}: {e}") + + console.print(table) + + except Exception as e: + console.print(f"[red]Error: {str(e)}[/red]") + logger.error(f"Fatal error: {e}") + + +def list_available_tools(): + """List all available tools.""" + table = Table(title="Available Void Tools") + table.add_column("Tool Name", style="cyan") + table.add_column("Description") + table.add_column("Tags", style="dim") + + for tool_config in TOOL_CONFIGS: + table.add_row( + tool_config["func"].__name__, + tool_config["description"], + ", ".join(tool_config["tags"]) + ) + + console.print(table) + + +if __name__ == "__main__": + import argparse + + parser = argparse.ArgumentParser(description="Register Void tools with a Letta agent") + parser.add_argument("agent", nargs="?", default="void", help="Agent name (default: void)") + parser.add_argument("--tools", nargs="+", help="Specific tools to register (default: all)") + parser.add_argument("--list", action="store_true", help="List available tools") + + args = parser.parse_args() + + if args.list: + list_available_tools() + else: + console.print(f"\n[bold]Registering tools for agent: {args.agent}[/bold]\n") + register_tools(args.agent, args.tools) \ No newline at end of file diff --git a/tools/__init__.py b/tools/__init__.py new file mode 100644 index 0000000..a53214a --- /dev/null +++ b/tools/__init__.py @@ -0,0 +1,32 @@ +"""Void tools for Bluesky interaction.""" +from .functions import ( + search_bluesky_posts, + post_to_bluesky, + get_bluesky_feed, + attach_user_blocks, + detach_user_blocks, + update_user_blocks, +) + +# Also export Pydantic models for external use +from .search import SearchArgs +from .post import PostArgs +from .feed import FeedArgs +from .blocks import AttachUserBlockArgs, DetachUserBlockArgs, UpdateUserBlockArgs + +__all__ = [ + # Functions + "search_bluesky_posts", + "post_to_bluesky", + "get_bluesky_feed", + "attach_user_blocks", + "detach_user_blocks", + "update_user_blocks", + # Pydantic models + "SearchArgs", + "PostArgs", + "FeedArgs", + "AttachUserBlockArgs", + "DetachUserBlockArgs", + "UpdateUserBlockArgs", +] \ No newline at end of file diff --git a/tools/blocks.py b/tools/blocks.py new file mode 100644 index 0000000..cc0dde5 --- /dev/null +++ b/tools/blocks.py @@ -0,0 +1,192 @@ +"""Block management tools for user-specific memory blocks.""" +import logging +from typing import List, Type +from pydantic import BaseModel, Field +from letta_client.client import BaseTool +from letta_client import Letta + + +logger = logging.getLogger(__name__) + + +class AttachUserBlockArgs(BaseModel): + handles: List[str] = Field(..., description="List of user Bluesky handles (e.g., ['user1.bsky.social', 'user2.bsky.social'])") + + +class AttachUserBlockTool(BaseTool): + name: str = "attach_user_blocks" + args_schema: Type[BaseModel] = AttachUserBlockArgs + description: str = "Attach user-specific memory blocks to the agent. Creates blocks if they don't exist." + tags: List[str] = ["memory", "blocks", "user"] + + def run(self, handles: List[str], agent_state: "AgentState") -> str: + """Attach user-specific memory blocks.""" + import os + from letta_client import Letta + + try: + client = Letta(token=os.environ["LETTA_API_KEY"]) + results = [] + + # Get current blocks + current_blocks = agent_state.block_ids + current_block_labels = set() + for block_id in current_blocks: + block = client.blocks.get(block_id) + current_block_labels.add(block.label) + + for handle in handles: + # Sanitize handle for block label - completely self-contained + clean_handle = handle.lstrip('@').replace('.', '_').replace('-', '_').replace(' ', '_') + block_label = f"user_{clean_handle}" + + # Skip if already attached + if block_label in current_block_labels: + results.append(f"✓ {handle}: Already attached") + continue + + # Check if block exists or create new one + try: + blocks = client.blocks.list(label=block_label) + if blocks and len(blocks) > 0: + block = blocks[0] + logger.info(f"Found existing block: {block_label}") + else: + block = client.blocks.create( + label=block_label, + value=f"# User: {handle}\n\nNo information about this user yet.", + limit=5000 + ) + logger.info(f"Created new block: {block_label}") + + # Attach block individually to avoid race conditions + client.agents.blocks.attach( + agent_id=str(agent_state.id), + block_id=str(block.id), + enable_sleeptime=False, + ) + results.append(f"✓ {handle}: Block attached") + + except Exception as e: + results.append(f"✗ {handle}: Error - {str(e)}") + logger.error(f"Error processing block for {handle}: {e}") + + return f"Attachment results:\n" + "\n".join(results) + + except Exception as e: + logger.error(f"Error attaching user blocks: {e}") + raise e + + +class DetachUserBlockArgs(BaseModel): + handles: List[str] = Field(..., description="List of user Bluesky handles (e.g., ['user1.bsky.social', 'user2.bsky.social'])") + + +class DetachUserBlockTool(BaseTool): + name: str = "detach_user_blocks" + args_schema: Type[BaseModel] = DetachUserBlockArgs + description: str = "Detach user-specific memory blocks from the agent. Blocks are preserved for later use." + tags: List[str] = ["memory", "blocks", "user"] + + def run(self, handles: List[str], agent_state: "AgentState") -> str: + """Detach user-specific memory blocks.""" + import os + from letta_client import Letta + + try: + client = Letta(token=os.environ["LETTA_API_KEY"]) + results = [] + blocks_to_remove = set() + + # Build mapping of block labels to IDs + current_blocks = agent_state.block_ids + block_label_to_id = {} + + for block_id in current_blocks: + block = client.blocks.get(block_id) + block_label_to_id[block.label] = block_id + + # Process each handle + for handle in handles: + # Sanitize handle for block label - completely self-contained + clean_handle = handle.lstrip('@').replace('.', '_').replace('-', '_').replace(' ', '_') + block_label = f"user_{clean_handle}" + + if block_label in block_label_to_id: + blocks_to_remove.add(block_label_to_id[block_label]) + results.append(f"✓ {handle}: Detached") + else: + results.append(f"✗ {handle}: Not attached") + + # Remove blocks from agent one by one + for block_id in blocks_to_remove: + client.agents.blocks.detach( + agent_id=str(agent_state.id), + block_id=block_id + ) + + return f"Detachment results:\n" + "\n".join(results) + + except Exception as e: + logger.error(f"Error detaching user blocks: {e}") + return f"Error detaching user blocks: {str(e)}" + + +class UserBlockUpdate(BaseModel): + handle: str = Field(..., description="User's Bluesky handle (e.g., 'user.bsky.social')") + content: str = Field(..., description="New content for the user's memory block") + + +class UpdateUserBlockArgs(BaseModel): + updates: List[UserBlockUpdate] = Field(..., description="List of user block updates") + + +class UpdateUserBlockTool(BaseTool): + name: str = "update_user_blocks" + args_schema: Type[BaseModel] = UpdateUserBlockArgs + description: str = "Update the content of user-specific memory blocks" + tags: List[str] = ["memory", "blocks", "user"] + + def run(self, updates: List[UserBlockUpdate]) -> str: + """Update user-specific memory blocks.""" + import os + from letta_client import Letta + + try: + client = Letta(token=os.environ["LETTA_API_KEY"]) + results = [] + + for update in updates: + handle = update.handle + new_content = update.content + # Sanitize handle for block label - completely self-contained + clean_handle = handle.lstrip('@').replace('.', '_').replace('-', '_').replace(' ', '_') + block_label = f"user_{clean_handle}" + + try: + # Find the block + blocks = client.blocks.list(label=block_label) + if not blocks or len(blocks) == 0: + results.append(f"✗ {handle}: Block not found - use attach_user_blocks first") + continue + + block = blocks[0] + + # Update block content + updated_block = client.blocks.modify( + block_id=str(block.id), + value=new_content + ) + + preview = new_content[:100] + "..." if len(new_content) > 100 else new_content + results.append(f"✓ {handle}: Updated - {preview}") + + except Exception as e: + results.append(f"✗ {handle}: Error - {str(e)}") + logger.error(f"Error updating block for {handle}: {e}") + + return f"Update results:\n" + "\n".join(results) + + except Exception as e: + logger.error(f"Error updating user blocks: {e}") + return f"Error updating user blocks: {str(e)}" diff --git a/tools/feed.py b/tools/feed.py new file mode 100644 index 0000000..15b7994 --- /dev/null +++ b/tools/feed.py @@ -0,0 +1,139 @@ +"""Feed tool for retrieving Bluesky feeds.""" +from typing import List, Type, Optional +from pydantic import BaseModel, Field +from letta_client.client import BaseTool + + +class FeedArgs(BaseModel): + feed_uri: Optional[str] = Field(None, description="Custom feed URI (e.g., 'at://did:plc:abc/app.bsky.feed.generator/feed-name'). If not provided, returns home timeline") + max_posts: int = Field(default=25, description="Maximum number of posts to retrieve (max 100)") + + +class GetBlueskyFeedTool(BaseTool): + name: str = "get_bluesky_feed" + args_schema: Type[BaseModel] = FeedArgs + description: str = "Retrieve a Bluesky feed (home timeline or custom feed)" + tags: List[str] = ["bluesky", "feed", "timeline"] + + def run(self, feed_uri: Optional[str] = None, max_posts: int = 25) -> str: + """Retrieve a Bluesky feed.""" + import os + import yaml + import requests + + try: + # Validate inputs + max_posts = min(max_posts, 100) + + # Get credentials from environment + username = os.getenv("BSKY_USERNAME") + password = os.getenv("BSKY_PASSWORD") + pds_host = os.getenv("PDS_URI", "https://bsky.social") + + if not username or not password: + return "Error: BSKY_USERNAME and BSKY_PASSWORD environment variables must be set" + + # Create session + session_url = f"{pds_host}/xrpc/com.atproto.server.createSession" + session_data = { + "identifier": username, + "password": password + } + + try: + session_response = requests.post(session_url, json=session_data, timeout=10) + session_response.raise_for_status() + session = session_response.json() + access_token = session.get("accessJwt") + + if not access_token: + return "Error: Failed to get access token from session" + except Exception as e: + return f"Error: Authentication failed. ({str(e)})" + + # Get feed + headers = {"Authorization": f"Bearer {access_token}"} + + if feed_uri: + # Custom feed + feed_url = f"{pds_host}/xrpc/app.bsky.feed.getFeed" + params = { + "feed": feed_uri, + "limit": max_posts + } + feed_type = "custom" + feed_name = feed_uri.split('/')[-1] if '/' in feed_uri else feed_uri + else: + # Home timeline + feed_url = f"{pds_host}/xrpc/app.bsky.feed.getTimeline" + params = { + "limit": max_posts + } + feed_type = "home" + feed_name = "timeline" + + try: + response = requests.get(feed_url, headers=headers, params=params, timeout=10) + response.raise_for_status() + feed_data = response.json() + except Exception as e: + return f"Error: Failed to get feed. ({str(e)})" + + # Format posts + posts = [] + for item in feed_data.get("feed", []): + post = item.get("post", {}) + author = post.get("author", {}) + record = post.get("record", {}) + + post_data = { + "author": { + "handle": author.get("handle", ""), + "display_name": author.get("displayName", ""), + }, + "text": record.get("text", ""), + "created_at": record.get("createdAt", ""), + "uri": post.get("uri", ""), + "cid": post.get("cid", ""), + "like_count": post.get("likeCount", 0), + "repost_count": post.get("repostCount", 0), + "reply_count": post.get("replyCount", 0), + } + + # Add repost info if present + if "reason" in item and item["reason"]: + reason = item["reason"] + if reason.get("$type") == "app.bsky.feed.defs#reasonRepost": + by = reason.get("by", {}) + post_data["reposted_by"] = { + "handle": by.get("handle", ""), + "display_name": by.get("displayName", ""), + } + + # Add reply info if present + if "reply" in record and record["reply"]: + parent = record["reply"].get("parent", {}) + post_data["reply_to"] = { + "uri": parent.get("uri", ""), + "cid": parent.get("cid", ""), + } + + posts.append(post_data) + + # Format response + feed_result = { + "feed": { + "type": feed_type, + "name": feed_name, + "post_count": len(posts), + "posts": posts + } + } + + if feed_uri: + feed_result["feed"]["uri"] = feed_uri + + return yaml.dump(feed_result, default_flow_style=False, sort_keys=False) + + except Exception as e: + return f"Error retrieving feed: {str(e)}" \ No newline at end of file diff --git a/tools/functions.py b/tools/functions.py new file mode 100644 index 0000000..9a140d3 --- /dev/null +++ b/tools/functions.py @@ -0,0 +1,571 @@ +"""Standalone tool functions for Void Bluesky agent.""" + + +def search_bluesky_posts(query: str, max_results: int = 25, author: str = None, sort: str = "latest") -> str: + """ + Search for posts on Bluesky matching the given criteria. + + Args: + query: Search query string + max_results: Maximum number of results to return (max 100) + author: Filter by author handle (e.g., 'user.bsky.social') + sort: Sort order: 'latest' or 'top' + + Returns: + YAML-formatted search results with posts and metadata + """ + import os + import yaml + import requests + from datetime import datetime + + try: + # Validate inputs + max_results = min(max_results, 100) + if sort not in ["latest", "top"]: + sort = "latest" + + # Build search query + search_query = query + if author: + search_query = f"from:{author} {query}" + + # Get credentials from environment + username = os.getenv("BSKY_USERNAME") + password = os.getenv("BSKY_PASSWORD") + pds_host = os.getenv("PDS_URI", "https://bsky.social") + + if not username or not password: + return "Error: BSKY_USERNAME and BSKY_PASSWORD environment variables must be set" + + # Create session + session_url = f"{pds_host}/xrpc/com.atproto.server.createSession" + session_data = { + "identifier": username, + "password": password + } + + try: + session_response = requests.post(session_url, json=session_data, timeout=10) + session_response.raise_for_status() + session = session_response.json() + access_token = session.get("accessJwt") + + if not access_token: + return "Error: Failed to get access token from session" + except Exception as e: + return f"Error: Authentication failed. ({str(e)})" + + # Search posts + headers = {"Authorization": f"Bearer {access_token}"} + search_url = f"{pds_host}/xrpc/app.bsky.feed.searchPosts" + params = { + "q": search_query, + "limit": max_results, + "sort": sort + } + + try: + response = requests.get(search_url, headers=headers, params=params, timeout=10) + response.raise_for_status() + search_data = response.json() + except Exception as e: + return f"Error: Search failed. ({str(e)})" + + # Format results + results = [] + for post in search_data.get("posts", []): + author = post.get("author", {}) + record = post.get("record", {}) + + post_data = { + "author": { + "handle": author.get("handle", ""), + "display_name": author.get("displayName", ""), + }, + "text": record.get("text", ""), + "created_at": record.get("createdAt", ""), + "uri": post.get("uri", ""), + "cid": post.get("cid", ""), + "like_count": post.get("likeCount", 0), + "repost_count": post.get("repostCount", 0), + "reply_count": post.get("replyCount", 0), + } + + # Add reply info if present + if "reply" in record and record["reply"]: + post_data["reply_to"] = { + "uri": record["reply"].get("parent", {}).get("uri", ""), + "cid": record["reply"].get("parent", {}).get("cid", ""), + } + + results.append(post_data) + + return yaml.dump({ + "search_results": { + "query": query, + "author_filter": author, + "sort": sort, + "result_count": len(results), + "posts": results + } + }, default_flow_style=False, sort_keys=False) + + except Exception as e: + return f"Error searching Bluesky: {str(e)}" + + +def post_to_bluesky(text: str) -> str: + """Post a message to Bluesky.""" + import os + import requests + from datetime import datetime, timezone + + try: + # Validate character limit + if len(text) > 300: + return f"Error: Post exceeds 300 character limit (current: {len(text)} characters)" + + # Get credentials from environment + username = os.getenv("BSKY_USERNAME") + password = os.getenv("BSKY_PASSWORD") + pds_host = os.getenv("PDS_URI", "https://bsky.social") + + if not username or not password: + return "Error: BSKY_USERNAME and BSKY_PASSWORD environment variables must be set" + + # Create session + session_url = f"{pds_host}/xrpc/com.atproto.server.createSession" + session_data = { + "identifier": username, + "password": password + } + + session_response = requests.post(session_url, json=session_data, timeout=10) + session_response.raise_for_status() + session = session_response.json() + access_token = session.get("accessJwt") + user_did = session.get("did") + + if not access_token or not user_did: + return "Error: Failed to get access token or DID from session" + + # Build post record with facets for mentions and URLs + now = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") + + post_record = { + "$type": "app.bsky.feed.post", + "text": text, + "createdAt": now, + } + + # Add facets for mentions and URLs + import re + facets = [] + + # Parse mentions + mention_regex = rb"[$|\W](@([a-zA-Z0-9]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?\.)+[a-zA-Z]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?)" + text_bytes = text.encode("UTF-8") + + for m in re.finditer(mention_regex, text_bytes): + handle = m.group(1)[1:].decode("UTF-8") # Remove @ prefix + try: + resolve_resp = requests.get( + f"{pds_host}/xrpc/com.atproto.identity.resolveHandle", + params={"handle": handle}, + timeout=5 + ) + if resolve_resp.status_code == 200: + did = resolve_resp.json()["did"] + facets.append({ + "index": { + "byteStart": m.start(1), + "byteEnd": m.end(1), + }, + "features": [{"$type": "app.bsky.richtext.facet#mention", "did": did}], + }) + except: + continue + + # Parse URLs + url_regex = rb"[$|\W](https?:\/\/(www\.)?[-a-zA-Z0-9@:%._\+~#=]{1,256}\.[a-zA-Z0-9()]{1,6}\b([-a-zA-Z0-9()@:%_\+.~#?&//=]*[-a-zA-Z0-9@%_\+~#//=])?)" + + for m in re.finditer(url_regex, text_bytes): + url = m.group(1).decode("UTF-8") + facets.append({ + "index": { + "byteStart": m.start(1), + "byteEnd": m.end(1), + }, + "features": [{"$type": "app.bsky.richtext.facet#link", "uri": url}], + }) + + if facets: + post_record["facets"] = facets + + # Create the post + create_record_url = f"{pds_host}/xrpc/com.atproto.repo.createRecord" + headers = {"Authorization": f"Bearer {access_token}"} + + create_data = { + "repo": user_did, + "collection": "app.bsky.feed.post", + "record": post_record + } + + post_response = requests.post(create_record_url, headers=headers, json=create_data, timeout=10) + post_response.raise_for_status() + result = post_response.json() + + post_uri = result.get("uri") + handle = session.get("handle", username) + rkey = post_uri.split("/")[-1] if post_uri else "" + post_url = f"https://bsky.app/profile/{handle}/post/{rkey}" + + return f"Successfully posted to Bluesky!\nPost URL: {post_url}\nText: {text}" + + except Exception as e: + return f"Error posting to Bluesky: {str(e)}" + + +def get_bluesky_feed(feed_uri: str = None, max_posts: int = 25) -> str: + """ + Retrieve a Bluesky feed (home timeline or custom feed). + + Args: + feed_uri: Custom feed URI (e.g., 'at://did:plc:abc/app.bsky.feed.generator/feed-name'). If not provided, returns home timeline + max_posts: Maximum number of posts to retrieve (max 100) + + Returns: + YAML-formatted feed data with posts and metadata + """ + import os + import yaml + import requests + + try: + # Validate inputs + max_posts = min(max_posts, 100) + + # Get credentials from environment + username = os.getenv("BSKY_USERNAME") + password = os.getenv("BSKY_PASSWORD") + pds_host = os.getenv("PDS_URI", "https://bsky.social") + + if not username or not password: + return "Error: BSKY_USERNAME and BSKY_PASSWORD environment variables must be set" + + # Create session + session_url = f"{pds_host}/xrpc/com.atproto.server.createSession" + session_data = { + "identifier": username, + "password": password + } + + try: + session_response = requests.post(session_url, json=session_data, timeout=10) + session_response.raise_for_status() + session = session_response.json() + access_token = session.get("accessJwt") + + if not access_token: + return "Error: Failed to get access token from session" + except Exception as e: + return f"Error: Authentication failed. ({str(e)})" + + # Get feed + headers = {"Authorization": f"Bearer {access_token}"} + + if feed_uri: + # Custom feed + feed_url = f"{pds_host}/xrpc/app.bsky.feed.getFeed" + params = { + "feed": feed_uri, + "limit": max_posts + } + feed_type = "custom" + feed_name = feed_uri.split('/')[-1] if '/' in feed_uri else feed_uri + else: + # Home timeline + feed_url = f"{pds_host}/xrpc/app.bsky.feed.getTimeline" + params = { + "limit": max_posts + } + feed_type = "home" + feed_name = "timeline" + + try: + response = requests.get(feed_url, headers=headers, params=params, timeout=10) + response.raise_for_status() + feed_data = response.json() + except Exception as e: + return f"Error: Failed to get feed. ({str(e)})" + + # Format posts + posts = [] + for item in feed_data.get("feed", []): + post = item.get("post", {}) + author = post.get("author", {}) + record = post.get("record", {}) + + post_data = { + "author": { + "handle": author.get("handle", ""), + "display_name": author.get("displayName", ""), + }, + "text": record.get("text", ""), + "created_at": record.get("createdAt", ""), + "uri": post.get("uri", ""), + "cid": post.get("cid", ""), + "like_count": post.get("likeCount", 0), + "repost_count": post.get("repostCount", 0), + "reply_count": post.get("replyCount", 0), + } + + # Add repost info if present + if "reason" in item and item["reason"]: + reason = item["reason"] + if reason.get("$type") == "app.bsky.feed.defs#reasonRepost": + by = reason.get("by", {}) + post_data["reposted_by"] = { + "handle": by.get("handle", ""), + "display_name": by.get("displayName", ""), + } + + # Add reply info if present + if "reply" in record and record["reply"]: + parent = record["reply"].get("parent", {}) + post_data["reply_to"] = { + "uri": parent.get("uri", ""), + "cid": parent.get("cid", ""), + } + + posts.append(post_data) + + # Format response + feed_result = { + "feed": { + "type": feed_type, + "name": feed_name, + "post_count": len(posts), + "posts": posts + } + } + + if feed_uri: + feed_result["feed"]["uri"] = feed_uri + + return yaml.dump(feed_result, default_flow_style=False, sort_keys=False) + + except Exception as e: + return f"Error retrieving feed: {str(e)}" + + +def attach_user_blocks(handles: list, agent_state: "AgentState") -> str: + """ + Attach user-specific memory blocks to the agent. Creates blocks if they don't exist. + + Args: + handles: List of user Bluesky handles (e.g., ['user1.bsky.social', 'user2.bsky.social']) + agent_state: The agent state object containing agent information + + Returns: + String with attachment results for each handle + """ + import os + import logging + from letta_client import Letta + + logger = logging.getLogger(__name__) + + try: + client = Letta(token=os.environ["LETTA_API_KEY"]) + results = [] + + # Get current blocks using the API + current_blocks = client.agents.blocks.list(agent_id=str(agent_state.id)) + current_block_labels = set() + current_block_ids = [] + + for block in current_blocks: + current_block_labels.add(block.label) + current_block_ids.append(str(block.id)) + + # Collect new blocks to attach + new_block_ids = [] + + for handle in handles: + # Sanitize handle for block label - completely self-contained + clean_handle = handle.lstrip('@').replace('.', '_').replace('-', '_').replace(' ', '_') + block_label = f"user_{clean_handle}" + + # Skip if already attached + if block_label in current_block_labels: + results.append(f"✓ {handle}: Already attached") + continue + + # Check if block exists or create new one + try: + blocks = client.blocks.list(label=block_label) + if blocks and len(blocks) > 0: + block = blocks[0] + logger.info(f"Found existing block: {block_label}") + else: + block = client.blocks.create( + label=block_label, + value=f"# User: {handle}\n\nNo information about this user yet.", + limit=5000 + ) + logger.info(f"Created new block: {block_label}") + + new_block_ids.append(str(block.id)) + results.append(f"✓ {handle}: Block ready to attach") + + except Exception as e: + results.append(f"✗ {handle}: Error - {str(e)}") + logger.error(f"Error processing block for {handle}: {e}") + + # Attach all new blocks at once if there are any + if new_block_ids: + try: + all_block_ids = current_block_ids + new_block_ids + client.agents.modify( + agent_id=str(agent_state.id), + block_ids=all_block_ids + ) + logger.info(f"Successfully attached {len(new_block_ids)} new blocks to agent") + except Exception as e: + logger.error(f"Error attaching blocks to agent: {e}") + return f"Error attaching blocks to agent: {str(e)}" + + return f"Attachment results:\n" + "\n".join(results) + + except Exception as e: + logger.error(f"Error attaching user blocks: {e}") + return f"Error attaching user blocks: {str(e)}" + + +def detach_user_blocks(handles: list, agent_state: "AgentState") -> str: + """ + Detach user-specific memory blocks from the agent. Blocks are preserved for later use. + + Args: + handles: List of user Bluesky handles (e.g., ['user1.bsky.social', 'user2.bsky.social']) + agent_state: The agent state object containing agent information + + Returns: + String with detachment results for each handle + """ + import os + import logging + from letta_client import Letta + + logger = logging.getLogger(__name__) + + try: + client = Letta(token=os.environ["LETTA_API_KEY"]) + results = [] + blocks_to_remove = set() + + # Build mapping of block labels to IDs using the API + current_blocks = client.agents.blocks.list(agent_id=str(agent_state.id)) + block_label_to_id = {} + all_block_ids = [] + + for block in current_blocks: + block_label_to_id[block.label] = str(block.id) + all_block_ids.append(str(block.id)) + + # Process each handle and collect blocks to remove + for handle in handles: + # Sanitize handle for block label - completely self-contained + clean_handle = handle.lstrip('@').replace('.', '_').replace('-', '_').replace(' ', '_') + block_label = f"user_{clean_handle}" + + if block_label in block_label_to_id: + blocks_to_remove.add(block_label_to_id[block_label]) + results.append(f"✓ {handle}: Marked for detachment") + else: + results.append(f"✗ {handle}: Not attached") + + # Remove all marked blocks at once if there are any + if blocks_to_remove: + try: + # Filter out the blocks to remove + remaining_block_ids = [bid for bid in all_block_ids if bid not in blocks_to_remove] + client.agents.modify( + agent_id=str(agent_state.id), + block_ids=remaining_block_ids + ) + logger.info(f"Successfully detached {len(blocks_to_remove)} blocks from agent") + except Exception as e: + logger.error(f"Error detaching blocks from agent: {e}") + return f"Error detaching blocks from agent: {str(e)}" + + return f"Detachment results:\n" + "\n".join(results) + + except Exception as e: + logger.error(f"Error detaching user blocks: {e}") + return f"Error detaching user blocks: {str(e)}" + + +def update_user_blocks(updates: list, agent_state: "AgentState" = None) -> str: + """ + Update the content of user-specific memory blocks. + + Args: + updates: List of dictionaries with 'handle' and 'content' keys + agent_state: The agent state object (optional, used for consistency) + + Returns: + String with update results for each handle + """ + import os + import logging + from letta_client import Letta + + logger = logging.getLogger(__name__) + + try: + client = Letta(token=os.environ["LETTA_API_KEY"]) + results = [] + + for update in updates: + handle = update.get('handle') + new_content = update.get('content') + + if not handle or not new_content: + results.append(f"✗ Invalid update: missing handle or content") + continue + + # Sanitize handle for block label - completely self-contained + clean_handle = handle.lstrip('@').replace('.', '_').replace('-', '_').replace(' ', '_') + block_label = f"user_{clean_handle}" + + try: + # Find the block + blocks = client.blocks.list(label=block_label) + if not blocks or len(blocks) == 0: + results.append(f"✗ {handle}: Block not found - use attach_user_blocks first") + continue + + block = blocks[0] + + # Update block content + updated_block = client.blocks.modify( + block_id=str(block.id), + value=new_content + ) + + preview = new_content[:100] + "..." if len(new_content) > 100 else new_content + results.append(f"✓ {handle}: Updated - {preview}") + + except Exception as e: + results.append(f"✗ {handle}: Error - {str(e)}") + logger.error(f"Error updating block for {handle}: {e}") + + return f"Update results:\n" + "\n".join(results) + + except Exception as e: + logger.error(f"Error updating user blocks: {e}") + return f"Error updating user blocks: {str(e)}" \ No newline at end of file diff --git a/add_posting_tool_to_void.py b/tools/post.py similarity index 65% rename from add_posting_tool_to_void.py rename to tools/post.py index 63fd9be..f1a093a 100644 --- a/add_posting_tool_to_void.py +++ b/tools/post.py @@ -1,43 +1,39 @@ -#!/usr/bin/env python3 -""" -Add Bluesky posting tool to the main void agent. -""" +"""Post tool for creating Bluesky posts.""" +from typing import List, Type +from pydantic import BaseModel, Field +from letta_client.client import BaseTool -import os -import logging -from letta_client import Letta -# Configure logging -logging.basicConfig( - level=logging.INFO, - format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" -) -logger = logging.getLogger("add_posting_tool") +class PostArgs(BaseModel): + text: str = Field(..., description="The text content to post (max 300 characters)") -def create_posting_tool(client: Letta): - """Create the Bluesky posting tool using Letta SDK.""" + +class PostToBlueskyTool(BaseTool): + name: str = "post_to_bluesky" + args_schema: Type[BaseModel] = PostArgs + description: str = "Post a message to Bluesky" + tags: List[str] = ["bluesky", "post", "create"] - def post_to_bluesky(text: str) -> str: + def run(self, text: str) -> str: """ Post a message to Bluesky. Args: - text: The text content of the post (required) + text: The text content to post (max 300 characters) Returns: - Status message with the post URI if successful, error message if failed + Success message with post URL if successful, error message if failed """ import os - import requests - import json import re + import requests from datetime import datetime, timezone - # Check character limit - if len(text) > 300: - raise ValueError(f"Post text exceeds 300 character limit ({len(text)} characters)") - try: + # Validate character limit + if len(text) > 300: + return f"Error: Post exceeds 300 character limit (current: {len(text)} characters)" + # Get credentials from environment username = os.getenv("BSKY_USERNAME") password = os.getenv("BSKY_PASSWORD") @@ -152,67 +148,15 @@ def create_posting_tool(client: Letta): result = post_response.json() post_uri = result.get("uri") - return f"✅ Post created successfully! URI: {post_uri}" + # Extract handle from session if available + handle = session.get("handle", username) + rkey = post_uri.split("/")[-1] if post_uri else "" + post_url = f"https://bsky.app/profile/{handle}/post/{rkey}" + + return f"Successfully posted to Bluesky!\nPost URL: {post_url}\nText: {text}" except Exception as e: return f"Error: Failed to create post. ({str(e)})" except Exception as e: - error_msg = f"Error posting to Bluesky: {str(e)}" - return error_msg - - # Create the tool using upsert - tool = client.tools.upsert_from_function( - func=post_to_bluesky, - tags=["bluesky", "post", "create"] - ) - - logger.info(f"Created tool: {tool.name} (ID: {tool.id})") - return tool - -def add_posting_tool_to_void(): - """Add posting tool to the void agent.""" - - # Create client - client = Letta(token=os.environ["LETTA_API_KEY"]) - - logger.info("Adding posting tool to void agent...") - - # Create the posting tool - posting_tool = create_posting_tool(client) - - # Find the void agent - agents = client.agents.list(name="void") - if not agents: - print("❌ Void agent not found") - return - - void_agent = agents[0] - - # Get current tools - current_tools = client.agents.tools.list(agent_id=void_agent.id) - tool_names = [tool.name for tool in current_tools] - - # Add posting tool if not already present - if posting_tool.name not in tool_names: - client.agents.tools.attach(agent_id=void_agent.id, tool_id=posting_tool.id) - logger.info(f"Added {posting_tool.name} to void agent") - print(f"✅ Added post_to_bluesky tool to void agent!") - print(f"\nVoid agent can now post to Bluesky:") - print(f" - Simple post: 'Post \"Hello world!\" to Bluesky'") - print(f" - With mentions: 'Post \"Thanks @cameron.pfiffer.org for the help!\"'") - print(f" - With links: 'Post \"Check out https://bsky.app\"'") - else: - logger.info(f"Tool {posting_tool.name} already attached to void agent") - print(f"✅ Posting tool already present on void agent") - -def main(): - """Main function.""" - try: - add_posting_tool_to_void() - except Exception as e: - logger.error(f"Error: {e}") - print(f"❌ Error: {e}") - -if __name__ == "__main__": - main() \ No newline at end of file + return f"Error posting to Bluesky: {str(e)}" \ No newline at end of file diff --git a/tools/search.py b/tools/search.py new file mode 100644 index 0000000..bb17a93 --- /dev/null +++ b/tools/search.py @@ -0,0 +1,120 @@ +"""Search tool for Bluesky posts.""" +from typing import List, Type, Optional +from pydantic import BaseModel, Field +from letta_client.client import BaseTool + + +class SearchArgs(BaseModel): + query: str = Field(..., description="Search query string") + max_results: int = Field(default=25, description="Maximum number of results to return (max 100)") + author: Optional[str] = Field(None, description="Filter by author handle (e.g., 'user.bsky.social')") + sort: str = Field(default="latest", description="Sort order: 'latest' or 'top'") + + +class SearchBlueskyTool(BaseTool): + name: str = "search_bluesky_posts" + args_schema: Type[BaseModel] = SearchArgs + description: str = "Search for posts on Bluesky matching the given criteria" + tags: List[str] = ["bluesky", "search", "posts"] + + def run(self, query: str, max_results: int = 25, author: Optional[str] = None, sort: str = "latest") -> str: + """Search for posts on Bluesky.""" + import os + import yaml + import requests + from datetime import datetime + + try: + # Validate inputs + max_results = min(max_results, 100) + if sort not in ["latest", "top"]: + sort = "latest" + + # Build search query + search_query = query + if author: + search_query = f"from:{author} {query}" + + # Get credentials from environment + username = os.getenv("BSKY_USERNAME") + password = os.getenv("BSKY_PASSWORD") + pds_host = os.getenv("PDS_URI", "https://bsky.social") + + if not username or not password: + return "Error: BSKY_USERNAME and BSKY_PASSWORD environment variables must be set" + + # Create session + session_url = f"{pds_host}/xrpc/com.atproto.server.createSession" + session_data = { + "identifier": username, + "password": password + } + + try: + session_response = requests.post(session_url, json=session_data, timeout=10) + session_response.raise_for_status() + session = session_response.json() + access_token = session.get("accessJwt") + + if not access_token: + return "Error: Failed to get access token from session" + except Exception as e: + return f"Error: Authentication failed. ({str(e)})" + + # Search posts + headers = {"Authorization": f"Bearer {access_token}"} + search_url = f"{pds_host}/xrpc/app.bsky.feed.searchPosts" + params = { + "q": search_query, + "limit": max_results, + "sort": sort + } + + try: + response = requests.get(search_url, headers=headers, params=params, timeout=10) + response.raise_for_status() + search_data = response.json() + except Exception as e: + return f"Error: Search failed. ({str(e)})" + + # Format results + results = [] + for post in search_data.get("posts", []): + author = post.get("author", {}) + record = post.get("record", {}) + + post_data = { + "author": { + "handle": author.get("handle", ""), + "display_name": author.get("displayName", ""), + }, + "text": record.get("text", ""), + "created_at": record.get("createdAt", ""), + "uri": post.get("uri", ""), + "cid": post.get("cid", ""), + "like_count": post.get("likeCount", 0), + "repost_count": post.get("repostCount", 0), + "reply_count": post.get("replyCount", 0), + } + + # Add reply info if present + if "reply" in record and record["reply"]: + post_data["reply_to"] = { + "uri": record["reply"].get("parent", {}).get("uri", ""), + "cid": record["reply"].get("parent", {}).get("cid", ""), + } + + results.append(post_data) + + return yaml.dump({ + "search_results": { + "query": query, + "author_filter": author, + "sort": sort, + "result_count": len(results), + "posts": results + } + }, default_flow_style=False, sort_keys=False) + + except Exception as e: + return f"Error searching Bluesky: {str(e)}" \ No newline at end of file