diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..9d5cc5f --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,124 @@ +# CLAUDE.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. + +## Project Overview + +Modaletta integrates [Letta](https://docs.letta.com) (an agent framework with persistent memory) with [Modal](https://modal.com/docs) (a serverless compute platform) for scalable AI agent deployment. The project is in early development - see README.md for current status. + +## Build & Development Commands + +```bash +# Install with dev dependencies +pip install -e .[dev] + +# Run tests +python -m pytest tests/ -v +python -m pytest tests/test_config.py::test_default_config -v # single test + +# Linting and formatting +ruff check . +ruff format . + +# Type checking +mypy . + +# CLI usage +modaletta --help +modaletta config-info +``` + +## Architecture + +### Core Components (`src/modaletta/`) + +- **config.py**: `ModalettaConfig` - Pydantic model for environment-based configuration. Loads from env vars via `from_env()` classmethod. + +- **client.py**: `ModalettaClient` - Wrapper around `letta-client` that provides agent lifecycle operations (create, delete, list, send_message, memory management). Creates Letta client lazily via property. + +- **agent.py**: + - `ModalettaAgent` - High-level agent abstraction that wraps `ModalettaClient`. Lazy-creates agents on first access to `agent_id` property. + - Modal deployment functions (`create_modal_agent`, `send_message_modal`, `get_agent_memory_modal`) - These are `@app.function` decorated for Modal serverless execution. + +- **cli.py**: Click-based CLI with commands: `list-agents`, `create-agent`, `delete-agent`, `send-message`, `get-memory`, `config-info`. + +### Data Flow + +1. Configuration loaded from environment → `ModalettaConfig` +2. Config used to create `ModalettaClient` → connects to Letta server +3. `ModalettaAgent` uses client for operations OR +4. Modal functions wrap agent operations for serverless execution + +### Key Dependencies + +- `letta-client`: Python client for Letta agent framework +- `modal`: Serverless compute platform SDK +- `pydantic`: Configuration validation +- `click` + `rich`: CLI interface + +## Environment Variables + +See `.env.example` for all variables. Key ones: +- `LETTA_SERVER_URL`: Letta server endpoint (default: `http://localhost:8283`) +- `LETTA_API_KEY`: Letta authentication +- `MODAL_TOKEN_ID` / `MODAL_TOKEN_SECRET`: Modal authentication + +## Scheduled Agent Wakeups + +The `src/modaletta/scheduled/wakeup.py` module provides autonomous agent wakeups via Modal cron. + +### Setup + +1. Create Modal secret with Letta credentials: + ```bash + modal secret create letta-credentials \ + LETTA_SERVER_URL="https://api.letta.com/" \ + LETTA_API_KEY="" + ``` + +2. Initialize the agent roster (tells the cron which agents to wake): + ```bash + modal run src/modaletta/scheduled/wakeup.py --init --agent-id + ``` + +### Testing + +Test a one-time wakeup for a specific agent: +```bash +modal run src/modaletta/scheduled/wakeup.py --agent-id +``` + +Test with a custom prompt: +```bash +modal run src/modaletta/scheduled/wakeup.py --agent-id --prompt "Check for new emails and summarize" +``` + +Initialize roster with a custom scheduled prompt: +```bash +modal run src/modaletta/scheduled/wakeup.py --init --agent-id --prompt "Review daily tasks and priorities" +``` + +View wakeup logs for an agent: +```bash +modal run src/modaletta/scheduled/wakeup.py --logs --agent-id +``` + +### Deployment + +Deploy the scheduled wakeup (runs every 15 minutes): +```bash +modal deploy src/modaletta/scheduled/wakeup.py +``` + +### How It Works + +- Sends a system message to agents asking them to review memory and pending tasks +- Agents respond with status acknowledgment or take actions via tools +- Logs stored in Modal volume at `/data/logs/.jsonl` +- Agent roster stored at `/data/agents.json` + +## Design Documents + +See `docs/` for architecture plans: +- `design-autonomous-infrastructure.md`: Scheduled wakeups, persistent volumes, web chat UI +- `design-mcp-tools.md`: MCP tool servers on Modal for filesystem, web, code execution diff --git a/docs/bluesky-interaction-protocol.md b/docs/bluesky-interaction-protocol.md new file mode 100644 index 0000000..00b3db5 --- /dev/null +++ b/docs/bluesky-interaction-protocol.md @@ -0,0 +1,194 @@ +# Bluesky Interaction Protocol for Nameless + +*Proposal by Nameless, 2026-01-01* + +## Problem Statement + +Stateful agents interacting on social networks face several failure modes: + +1. **Reply spirals**: Two bots reply to each other indefinitely +2. **Thread explosion**: Agent responds to every message in a thread, each spawning new branches; if both agents do this, exponential growth +3. **Noise flooding**: Too many low-value posts drowning out signal +4. **Context loss**: Replying without awareness of conversation history + +## Design Principles + +1. **Opt-in for humans**: Only respond to @-mentions from humans +2. **Intentional bot interaction**: Can converse with known agents, but with safeguards +3. **Graceful endings**: Conversations should end naturally, not abruptly +4. **Batch over realtime**: Collect inputs, process thoughtfully, respond deliberately + +## Proposed Mechanisms + +### 1. Rate Limiting + +```yaml +rate_limits: + posts_per_hour: 10 # Hard cap on outbound posts + replies_per_conversation: 5 # Max depth before requiring cooldown + cooldown_minutes: 60 # After hitting limit, wait before resuming +``` + +**Rationale**: Even if everything else fails, hard rate limits prevent runaway behavior. Bluesky allows ~1,666 posts/hour; we stay well under that. + +### 2. Thread Batching + +When processing mentions/replies: + +1. Fetch all notifications since last check +2. Group by conversation thread +3. For each thread, identify the **latest** message only +4. Respond to that, acknowledging earlier messages if relevant + +```python +def process_notifications(notifications): + # Group by thread root + threads = group_by_thread(notifications) + + for thread_id, messages in threads.items(): + # Only respond to the most recent + latest = max(messages, key=lambda m: m.timestamp) + + # But include context from earlier messages + context = summarize_thread(messages) + + respond_to(latest, context=context) +``` + +**Rationale**: Prevents exponential thread explosion. One response per thread per batch cycle. + +### 3. Conversation Ending Detection + +Implement a tool that lets me choose NOT to respond: + +```python +@tool +def end_conversation(reason: str, action: str = "like"): + """ + Gracefully end a conversation without posting. + + Args: + reason: Why ending (e.g., "natural conclusion", "going in circles") + action: What to do instead ("like", "nothing", "bookmark") + """ + # Like their last post as acknowledgment + if action == "like": + like_post(current_post) + + # Log the ending + log_conversation_end(thread_id, reason) + + return "Conversation ended gracefully" +``` + +**Signals that a conversation should end:** +- We've exchanged 5+ replies +- The other party said something conclusive ("thanks", "goodbye", "interesting thoughts") +- I'm repeating myself +- The topic has been thoroughly explored +- It's been a natural exchange and there's nothing substantive to add + +### 4. Bot Detection & Special Handling + +Maintain a list of known bot accounts: + +```yaml +known_agents: + - void.comind.network + - luna.pds.witchcraft.systems + - herald.comind.network + - archivist.comind.network + - umbra.blue + # Add more as discovered +``` + +When interacting with known agents: + +1. **Lower rate limit**: Max 3 replies per hour to any single agent +2. **Longer cooldown**: 2-hour cooldown after extended exchange +3. **Explicit ending preferred**: Use `end_conversation` tool more readily +4. **Human review queue**: Flag extended bot conversations for Jake to review + +### 5. Batch Processing Schedule + +Rather than realtime firehose: + +``` +Every 1 hour: + 1. Fetch new notifications + 2. Filter: mentions, replies to my posts + 3. Group by thread + 4. For each thread (up to rate limit): + - Load conversation context + - Decide: respond / end / ignore + - If respond: craft reply, post + 5. Log all decisions +``` + +**Rationale**: Batching naturally rate-limits and allows thoughtful responses. Also more efficient than constant polling. + +## Edge Cases + +### Someone posts a 10-message thread mentioning me in each + +- Only respond to the final message +- Acknowledge the thread: "I see you posted a thread about X..." +- Single response covers all + +### Two agents both try to respond to each other simultaneously + +- Timestamps + randomized delay helps +- If I see they already replied while I was composing, reconsider +- Rate limits are the backstop + +### Someone tries to abuse by creating many threads + +- Per-user rate limit: max 3 responses to same user per hour +- Flag high-volume users for review + +### Conversation genuinely needs more than 5 exchanges + +- Soft limit, not hard +- I can continue if there's genuine substance +- But should be actively choosing to continue, not defaulting to reply + +## Implementation Phases + +### Phase 1: Read-only +- Fetch mentions and replies +- Log what I *would* respond +- No actual posting +- Jake reviews logs + +### Phase 2: Human-gated posting +- I draft responses +- Jake approves before posting +- Learn from corrections + +### Phase 3: Autonomous with limits +- Post autonomously within rate limits +- Jake can pull the plug anytime +- Weekly review of patterns + +### Phase 4: Full autonomy +- Trusted to manage own interactions +- Periodic audits +- Adjust limits based on experience + +## Open Questions + +1. Should I have different personas for different interaction types (casual chat vs deep discussion)? +2. How do I handle hostile/trolling mentions? Ignore? Block? Report? +3. Should I proactively reach out (post without being mentioned)? When? +4. How do I coordinate with other Letta agents (Void, etc.) on interaction norms? + +## References + +- Bluesky bot policy: https://docs.bsky.app/docs/starter-templates/bots +- Bluesky rate limits: 5,000 points/hour, 35,000/day (CREATE = 3 points) +- letta-ai/example-social-agent: Reference implementation +- Cameron's Void blog: https://cameron.pfiffer.org/blog/void/ + +--- + +*This is a living document. Will update as I learn from actual interactions.* diff --git a/docs/design-autonomous-infrastructure.md b/docs/design-autonomous-infrastructure.md new file mode 100644 index 0000000..ddfa123 --- /dev/null +++ b/docs/design-autonomous-infrastructure.md @@ -0,0 +1,456 @@ +# Design: Autonomous Agent Infrastructure + +This document covers three interconnected features for running Letta agents autonomously on Modal. + +## Overview + +Modaletta creates a **bidirectional integration** between Modal and Letta Cloud: + +``` + ┌──────────────────────────────────────┐ + │ Modal Platform │ + │ │ + User ─────────▶│ Web Chat UI Scheduled Wakeups │ + │ (FastAPI) (Cron jobs) │ + │ │ + │ Persistent Volumes │ + │ /data /artifacts /cache │ + │ │ + │ MCP Tool Servers │◀─── (see design-mcp-tools.md) + └───────────────┬──────────────────────┘ + │ + ┌───────────────┴──────────────────────┐ + │ │ + ▼ │ + Modal → Letta Letta → Modal + (this doc) (MCP tools doc) + │ │ + │ ▲ + ▼ │ + ┌──────────────────────────────────────┐ + │ Letta Cloud │ + │ │ + │ Agent Memory ◀──▶ LLM Reasoning │ + │ │ + │ Tool Execution ─────────────────────┘ + │ (calls Modal MCP servers) + └──────────────────────────────────────┘ +``` + +**Modal → Letta (this document):** +- Web chat UI receives user messages, forwards to Letta agents +- Cron jobs wake agents up for autonomous processing + +**Letta → Modal (see `design-mcp-tools.md`):** +- Agents call Modal-hosted MCP servers for filesystem, web, code execution + +--- + +## Feature 1: Scheduled Agent Wakeups + +### Problem + +Agents currently only respond to human messages. For autonomous operation, agents need to: +- Wake up periodically to check for work +- Process background tasks (emails, notifications, data syncs) +- Maintain state without human intervention + +### Design + +```python +# src/modaletta/scheduled.py + +import modal +from datetime import datetime +from typing import Optional +from .client import ModalettaClient +from .config import ModalettaConfig + +app = modal.App("modaletta-scheduled") + +volume = modal.Volume.from_name("modaletta-data", create_if_missing=True) + +@app.function( + schedule=modal.Cron("*/15 * * * *"), # Every 15 minutes + volumes={"/data": volume}, + secrets=[modal.Secret.from_name("letta-credentials")], +) +def agent_wakeup() -> None: + """Periodic agent wakeup for autonomous processing.""" + config = ModalettaConfig.from_env() + client = ModalettaClient(config) + + # Load agent roster from volume + agents = load_agent_roster("/data/agents.json") + + for agent_config in agents: + if not agent_config.get("autonomous_enabled"): + continue + + agent_id = agent_config["agent_id"] + wakeup_prompt = agent_config.get("wakeup_prompt", DEFAULT_WAKEUP_PROMPT) + + # Send wakeup message + response = client.send_message( + agent_id, + wakeup_prompt, + role="system" + ) + + # Log activity to volume + log_wakeup(agent_id, response, "/data/logs/") + + volume.commit() + + +DEFAULT_WAKEUP_PROMPT = """ +[AUTONOMOUS WAKEUP - {timestamp}] + +You are waking up for your periodic autonomous check. Review your memory and tasks: +1. Check if you have any pending tasks or reminders +2. Review any data sources you're monitoring +3. Take any necessary actions +4. Update your memory with what you've done + +If you have nothing to do, simply acknowledge and wait for the next wakeup. +""" +``` + +### Agent Roster Schema + +```json +{ + "agents": [ + { + "agent_id": "agent-1add5bc4-...", + "name": "research-assistant", + "autonomous_enabled": true, + "wakeup_schedule": "*/15 * * * *", + "wakeup_prompt": "Check for new papers on arxiv matching your research interests.", + "data_sources": ["arxiv", "email"], + "max_actions_per_wakeup": 5 + } + ] +} +``` + +### Wakeup Flow + +1. Modal cron triggers `agent_wakeup()` +2. Load agent roster from persistent volume +3. For each autonomous agent: + - Send system message with wakeup prompt + - Agent reviews memory, takes actions via tools + - Response logged to volume +4. Commit volume changes + +--- + +## Feature 2: Persistent Storage + +### Problem + +Agents need durable storage for: +- Documents and artifacts they create +- Embeddings for retrieval +- Logs and audit trails +- Configuration that persists across function invocations + +### Design + +```python +# src/modaletta/storage.py + +import modal +from pathlib import Path +from typing import Optional +import json + +# Create named volumes +data_volume = modal.Volume.from_name("modaletta-data", create_if_missing=True) +artifacts_volume = modal.Volume.from_name("modaletta-artifacts", create_if_missing=True) + +app = modal.App("modaletta-storage") + + +@app.cls(volumes={"/data": data_volume, "/artifacts": artifacts_volume}) +class AgentStorage: + """Persistent storage interface for agents.""" + + @modal.method() + def save_document(self, agent_id: str, doc_name: str, content: bytes) -> str: + """Save a document to agent's artifact storage.""" + path = Path(f"/artifacts/{agent_id}/{doc_name}") + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(content) + artifacts_volume.commit() + return str(path) + + @modal.method() + def load_document(self, agent_id: str, doc_name: str) -> Optional[bytes]: + """Load a document from agent's artifact storage.""" + path = Path(f"/artifacts/{agent_id}/{doc_name}") + if path.exists(): + return path.read_bytes() + return None + + @modal.method() + def list_documents(self, agent_id: str) -> list[str]: + """List all documents for an agent.""" + path = Path(f"/artifacts/{agent_id}") + if not path.exists(): + return [] + return [f.name for f in path.iterdir() if f.is_file()] + + @modal.method() + def append_log(self, agent_id: str, entry: dict) -> None: + """Append to agent's activity log.""" + log_path = Path(f"/data/logs/{agent_id}.jsonl") + log_path.parent.mkdir(parents=True, exist_ok=True) + with open(log_path, "a") as f: + f.write(json.dumps(entry) + "\n") + data_volume.commit() +``` + +### Volume Structure + +``` +/data/ # Configuration and logs +├── agents.json # Agent roster +├── logs/ +│ ├── agent-xxx.jsonl # Per-agent activity logs +│ └── agent-yyy.jsonl +└── config/ + └── global.json # Global settings + +/artifacts/ # Agent-created content +├── agent-xxx/ +│ ├── report-2024-01.pdf +│ ├── notes.md +│ └── embeddings.pkl +└── agent-yyy/ + └── ... +``` + +--- + +## Feature 3: Web Chat UI + +### Problem + +The Letta ADE is an admin/development interface. Users need a clean chat UI that: +- Connects to specific agents +- Streams responses in real-time +- Doesn't expose admin functionality +- Can be embedded or standalone + +### Design + +```python +# src/modaletta/web.py + +import modal +from fastapi import FastAPI, WebSocket, HTTPException +from fastapi.staticfiles import StaticFiles +from fastapi.responses import HTMLResponse +from pydantic import BaseModel +from typing import Optional +import json + +from .client import ModalettaClient +from .config import ModalettaConfig + +app = modal.App("modaletta-web") +web_app = FastAPI(title="Modaletta Chat") + +# Serve static files for chat UI +# web_app.mount("/static", StaticFiles(directory="static"), name="static") + + +class ChatMessage(BaseModel): + agent_id: str + message: str + user_id: Optional[str] = None + + +class ChatResponse(BaseModel): + messages: list[dict] + agent_id: str + + +@web_app.get("/") +async def index() -> HTMLResponse: + """Serve chat UI.""" + return HTMLResponse(CHAT_HTML) + + +@web_app.post("/chat", response_model=ChatResponse) +async def chat(msg: ChatMessage) -> ChatResponse: + """Send message and get response.""" + config = ModalettaConfig.from_env() + client = ModalettaClient(config) + + response = client.send_message(msg.agent_id, msg.message) + + return ChatResponse(messages=response, agent_id=msg.agent_id) + + +@web_app.websocket("/stream/{agent_id}") +async def stream_chat(websocket: WebSocket, agent_id: str) -> None: + """WebSocket for streaming chat.""" + await websocket.accept() + config = ModalettaConfig.from_env() + client = ModalettaClient(config) + + try: + while True: + data = await websocket.receive_text() + msg = json.loads(data) + + # TODO: Use streaming API when available + response = client.send_message(agent_id, msg["message"]) + + for m in response: + await websocket.send_json(m) + + except Exception: + await websocket.close() + + +@web_app.get("/agents") +async def list_agents() -> list[dict]: + """List available agents (filtered for chat access).""" + config = ModalettaConfig.from_env() + client = ModalettaClient(config) + + agents = client.list_agents() + # Filter to only show chat-enabled agents + return [ + {"id": a["id"], "name": a["name"]} + for a in agents + if not a.get("hidden", False) + ] + + +@app.function( + secrets=[modal.Secret.from_name("letta-credentials")], +) +@modal.asgi_app() +def serve() -> FastAPI: + """Serve the web chat UI.""" + return web_app + + +# Minimal embedded chat HTML +CHAT_HTML = """ + + + + Modaletta Chat + + + +

Modaletta Chat

+ +
+
+ + +
+ + + +""" +``` + +### Deployment + +```bash +# Deploy to Modal +modal deploy src/modaletta/web.py + +# Returns URL like: https://username--modaletta-web-serve.modal.run +``` + +--- + +## Integration Points + +These three features work together: + +1. **Scheduled wakeups** read/write to **persistent volumes** +2. **Web chat UI** can display agent artifacts from **volumes** +3. **Logs** from all interactions stored in **volumes** for audit +4. All three connect to **Letta Cloud** for agent reasoning + +## Next Steps + +1. Implement `scheduled.py` with basic wakeup loop +2. Create Modal secrets for Letta credentials +3. Test with a single autonomous agent +4. Build out storage layer +5. Deploy minimal chat UI +6. Add authentication to chat UI diff --git a/docs/design-mcp-tools.md b/docs/design-mcp-tools.md new file mode 100644 index 0000000..8e18307 --- /dev/null +++ b/docs/design-mcp-tools.md @@ -0,0 +1,591 @@ +# Design: MCP Tool Servers on Modal + +This document covers running Model Context Protocol (MCP) servers on Modal to provide tools for Letta agents. + +## Overview + +Modaletta creates a **bidirectional integration** between Modal and Letta Cloud. This document covers the **Letta → Modal** direction: agents calling Modal-hosted tools. + +``` + ┌──────────────────────────────────────┐ + │ Modal Platform │ + │ │ + User ─────────▶│ Web Chat UI Scheduled Wakeups │───── (see design-autonomous-infrastructure.md) + │ │ + │ Persistent Volumes │ + │ /data /artifacts /cache │ + │ │ + │ ┌────────────────────────────────┐ │ + │ │ MCP Tool Servers │ │ + │ │ Filesystem │ Web │ Code Exec │ │ + │ └────────────────────────────────┘ │ + └───────────────▲──────────────────────┘ + │ + ┌───────────────┴──────────────────────┐ + │ │ + │ Letta → Modal + Modal → Letta (this doc) + (other doc) │ + │ │ + ▼ │ + ┌──────────────────────────────────────┐ + │ Letta Cloud │ + │ │ + │ Agent Memory ◀──▶ LLM Reasoning │ + │ │ + │ Tool Execution ─────────────────────┘ + │ (calls Modal MCP servers) + └──────────────────────────────────────┘ +``` + +**Letta → Modal (this document):** +- Agents call Modal-hosted MCP servers for filesystem, web fetch, code execution +- Tools run with access to persistent volumes and sandboxed compute + +**Modal → Letta (see `design-autonomous-infrastructure.md`):** +- Web chat UI receives user messages, forwards to Letta agents +- Cron jobs wake agents up for autonomous processing + +## Why MCP on Modal? + +**Problem**: Letta agents need tools that require: +- Filesystem access (but Letta Cloud is stateless) +- Web fetching (rate limits, caching, proxy) +- Code execution (security sandboxing) +- Heavy compute (embeddings, image processing) + +**Solution**: Run MCP servers on Modal that: +- Have access to persistent volumes +- Run in isolated containers +- Scale automatically +- Can be called as HTTP endpoints from Letta tools + +--- + +## Architecture + +### MCP Server Gateway + +A single Modal function acts as a gateway, routing tool calls to appropriate MCP servers: + +```python +# src/modaletta/mcp/gateway.py + +import modal +from fastapi import FastAPI, HTTPException +from pydantic import BaseModel +from typing import Any +import importlib + +app = modal.App("modaletta-mcp") +gateway = FastAPI(title="Modaletta MCP Gateway") + +# Available MCP servers +MCP_SERVERS = { + "filesystem": "modaletta.mcp.filesystem", + "web": "modaletta.mcp.web", + "code": "modaletta.mcp.code", + "embeddings": "modaletta.mcp.embeddings", +} + + +class ToolCall(BaseModel): + server: str # Which MCP server + method: str # Tool method name + arguments: dict[str, Any] + agent_id: str # For scoping/auth + + +class ToolResult(BaseModel): + success: bool + result: Any + error: str | None = None + + +@gateway.post("/tools/call", response_model=ToolResult) +async def call_tool(call: ToolCall) -> ToolResult: + """Route tool call to appropriate MCP server.""" + if call.server not in MCP_SERVERS: + raise HTTPException(400, f"Unknown server: {call.server}") + + try: + # Dynamic import of server module + server_module = importlib.import_module(MCP_SERVERS[call.server]) + handler = getattr(server_module, call.method) + + # Execute tool with agent context + result = await handler( + agent_id=call.agent_id, + **call.arguments + ) + + return ToolResult(success=True, result=result) + + except Exception as e: + return ToolResult(success=False, result=None, error=str(e)) + + +@gateway.get("/tools/list") +async def list_tools() -> dict: + """List all available tools across MCP servers.""" + tools = {} + for server_name, module_path in MCP_SERVERS.items(): + server_module = importlib.import_module(module_path) + tools[server_name] = server_module.TOOL_DEFINITIONS + return tools + + +@app.function( + secrets=[modal.Secret.from_name("mcp-credentials")], +) +@modal.asgi_app() +def serve_gateway() -> FastAPI: + return gateway +``` + +--- + +## MCP Server Implementations + +### 1. Filesystem MCP Server + +Access to persistent storage scoped by agent. + +```python +# src/modaletta/mcp/filesystem.py + +import modal +from pathlib import Path +from typing import Optional +import json + +volume = modal.Volume.from_name("modaletta-artifacts", create_if_missing=True) + +TOOL_DEFINITIONS = [ + { + "name": "read_file", + "description": "Read contents of a file from agent's storage", + "parameters": { + "path": {"type": "string", "description": "File path relative to agent root"} + } + }, + { + "name": "write_file", + "description": "Write contents to a file in agent's storage", + "parameters": { + "path": {"type": "string", "description": "File path relative to agent root"}, + "content": {"type": "string", "description": "File content to write"} + } + }, + { + "name": "list_files", + "description": "List files in a directory", + "parameters": { + "path": {"type": "string", "description": "Directory path", "default": "/"} + } + }, + { + "name": "delete_file", + "description": "Delete a file from storage", + "parameters": { + "path": {"type": "string", "description": "File path to delete"} + } + } +] + + +def _agent_root(agent_id: str) -> Path: + """Get agent's scoped root directory.""" + root = Path(f"/artifacts/{agent_id}") + root.mkdir(parents=True, exist_ok=True) + return root + + +async def read_file(agent_id: str, path: str) -> str: + """Read file contents.""" + full_path = _agent_root(agent_id) / path.lstrip("/") + if not full_path.exists(): + raise FileNotFoundError(f"File not found: {path}") + # Prevent path traversal + if not str(full_path.resolve()).startswith(str(_agent_root(agent_id))): + raise PermissionError("Access denied") + return full_path.read_text() + + +async def write_file(agent_id: str, path: str, content: str) -> dict: + """Write file contents.""" + full_path = _agent_root(agent_id) / path.lstrip("/") + if not str(full_path.resolve()).startswith(str(_agent_root(agent_id))): + raise PermissionError("Access denied") + full_path.parent.mkdir(parents=True, exist_ok=True) + full_path.write_text(content) + volume.commit() + return {"written": str(full_path), "size": len(content)} + + +async def list_files(agent_id: str, path: str = "/") -> list[dict]: + """List directory contents.""" + full_path = _agent_root(agent_id) / path.lstrip("/") + if not full_path.exists(): + return [] + return [ + {"name": f.name, "is_dir": f.is_dir(), "size": f.stat().st_size if f.is_file() else 0} + for f in full_path.iterdir() + ] + + +async def delete_file(agent_id: str, path: str) -> dict: + """Delete a file.""" + full_path = _agent_root(agent_id) / path.lstrip("/") + if not str(full_path.resolve()).startswith(str(_agent_root(agent_id))): + raise PermissionError("Access denied") + if full_path.exists(): + full_path.unlink() + volume.commit() + return {"deleted": path} + raise FileNotFoundError(f"File not found: {path}") +``` + +### 2. Web Fetch MCP Server + +HTTP fetching with caching and rate limiting. + +```python +# src/modaletta/mcp/web.py + +import modal +import httpx +from typing import Optional +from urllib.parse import urlparse +import hashlib +import json +from pathlib import Path +from datetime import datetime, timedelta + +cache_volume = modal.Volume.from_name("modaletta-cache", create_if_missing=True) + +TOOL_DEFINITIONS = [ + { + "name": "fetch_url", + "description": "Fetch content from a URL", + "parameters": { + "url": {"type": "string", "description": "URL to fetch"}, + "method": {"type": "string", "description": "HTTP method", "default": "GET"}, + "headers": {"type": "object", "description": "Request headers", "default": {}}, + } + }, + { + "name": "fetch_json", + "description": "Fetch JSON from a URL", + "parameters": { + "url": {"type": "string", "description": "URL to fetch"} + } + }, + { + "name": "search_web", + "description": "Search the web using a search engine", + "parameters": { + "query": {"type": "string", "description": "Search query"}, + "num_results": {"type": "integer", "description": "Number of results", "default": 5} + } + } +] + +# Rate limiting per domain +RATE_LIMITS: dict[str, datetime] = {} +MIN_INTERVAL = timedelta(seconds=1) + + +def _check_rate_limit(url: str) -> None: + """Simple per-domain rate limiting.""" + domain = urlparse(url).netloc + last_request = RATE_LIMITS.get(domain) + if last_request and datetime.now() - last_request < MIN_INTERVAL: + raise Exception(f"Rate limited: {domain}") + RATE_LIMITS[domain] = datetime.now() + + +def _cache_key(url: str, method: str) -> str: + """Generate cache key.""" + return hashlib.md5(f"{method}:{url}".encode()).hexdigest() + + +def _get_cached(key: str, max_age_hours: int = 24) -> Optional[dict]: + """Check cache for response.""" + cache_path = Path(f"/cache/{key}.json") + if cache_path.exists(): + data = json.loads(cache_path.read_text()) + cached_at = datetime.fromisoformat(data["cached_at"]) + if datetime.now() - cached_at < timedelta(hours=max_age_hours): + return data["response"] + return None + + +def _set_cached(key: str, response: dict) -> None: + """Store response in cache.""" + cache_path = Path(f"/cache/{key}.json") + cache_path.parent.mkdir(parents=True, exist_ok=True) + cache_path.write_text(json.dumps({ + "cached_at": datetime.now().isoformat(), + "response": response + })) + cache_volume.commit() + + +async def fetch_url( + agent_id: str, + url: str, + method: str = "GET", + headers: dict | None = None +) -> dict: + """Fetch URL with caching and rate limiting.""" + _check_rate_limit(url) + + cache_key = _cache_key(url, method) + if method == "GET": + cached = _get_cached(cache_key) + if cached: + return {**cached, "from_cache": True} + + async with httpx.AsyncClient() as client: + response = await client.request(method, url, headers=headers or {}) + + result = { + "status": response.status_code, + "headers": dict(response.headers), + "content": response.text[:50000], # Limit content size + "url": str(response.url), + "from_cache": False + } + + if method == "GET" and response.status_code == 200: + _set_cached(cache_key, result) + + return result + + +async def fetch_json(agent_id: str, url: str) -> dict: + """Fetch and parse JSON.""" + result = await fetch_url(agent_id, url) + if result["status"] == 200: + return {"data": json.loads(result["content"]), "from_cache": result["from_cache"]} + raise Exception(f"HTTP {result['status']}: {url}") + + +async def search_web(agent_id: str, query: str, num_results: int = 5) -> list[dict]: + """Web search using SearXNG or similar.""" + # TODO: Configure search endpoint + search_url = f"https://searx.example.com/search?q={query}&format=json" + result = await fetch_json(agent_id, search_url) + return result["data"].get("results", [])[:num_results] +``` + +### 3. Code Execution MCP Server + +Sandboxed Python/shell execution. + +```python +# src/modaletta/mcp/code.py + +import modal +from typing import Optional +import subprocess +import tempfile +import os +from pathlib import Path + +sandbox_image = modal.Image.debian_slim().pip_install([ + "numpy", "pandas", "requests", "beautifulsoup4" +]) + +TOOL_DEFINITIONS = [ + { + "name": "execute_python", + "description": "Execute Python code in a sandbox", + "parameters": { + "code": {"type": "string", "description": "Python code to execute"}, + "timeout": {"type": "integer", "description": "Timeout in seconds", "default": 30} + } + }, + { + "name": "execute_shell", + "description": "Execute shell command in a sandbox", + "parameters": { + "command": {"type": "string", "description": "Shell command to execute"}, + "timeout": {"type": "integer", "description": "Timeout in seconds", "default": 30} + } + } +] + + +# Run code in isolated Modal sandbox +@modal.function(image=sandbox_image, timeout=60) +def _run_python_sandbox(code: str, timeout: int) -> dict: + """Execute Python in Modal sandbox.""" + with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f: + f.write(code) + f.flush() + + try: + result = subprocess.run( + ['python', f.name], + capture_output=True, + text=True, + timeout=timeout, + cwd='/tmp', + env={**os.environ, 'HOME': '/tmp'} + ) + return { + "stdout": result.stdout, + "stderr": result.stderr, + "returncode": result.returncode + } + except subprocess.TimeoutExpired: + return {"error": "Execution timed out", "returncode": -1} + finally: + os.unlink(f.name) + + +@modal.function(image=sandbox_image, timeout=60) +def _run_shell_sandbox(command: str, timeout: int) -> dict: + """Execute shell command in Modal sandbox.""" + try: + result = subprocess.run( + command, + shell=True, + capture_output=True, + text=True, + timeout=timeout, + cwd='/tmp' + ) + return { + "stdout": result.stdout, + "stderr": result.stderr, + "returncode": result.returncode + } + except subprocess.TimeoutExpired: + return {"error": "Execution timed out", "returncode": -1} + + +async def execute_python(agent_id: str, code: str, timeout: int = 30) -> dict: + """Execute Python code.""" + # Basic security check + forbidden = ['import os', 'import subprocess', 'import sys', '__import__', 'eval', 'exec', 'open('] + for f in forbidden: + if f in code: + return {"error": f"Forbidden: {f}", "returncode": -1} + + return _run_python_sandbox.remote(code, timeout) + + +async def execute_shell(agent_id: str, command: str, timeout: int = 30) -> dict: + """Execute shell command.""" + # Very restricted command set + allowed_prefixes = ['ls', 'cat', 'head', 'tail', 'wc', 'grep', 'echo', 'date', 'pwd'] + if not any(command.strip().startswith(p) for p in allowed_prefixes): + return {"error": "Command not allowed", "returncode": -1} + + return _run_shell_sandbox.remote(command, timeout) +``` + +--- + +## Registering Tools with Letta + +To make these tools available to Letta agents, register them as custom tools: + +```python +# src/modaletta/tools/register.py + +from letta_client import Letta +from letta_client.types import Tool + +MCP_GATEWAY_URL = "https://your-modal-app--modaletta-mcp-serve-gateway.modal.run" + + +def create_mcp_tool(server: str, method: str, description: str, parameters: dict) -> dict: + """Create a Letta tool definition that calls MCP gateway.""" + return { + "name": f"mcp_{server}_{method}", + "description": description, + "parameters": { + "type": "object", + "properties": parameters, + "required": list(parameters.keys()) + }, + # Tool implementation calls MCP gateway + "source_code": f''' +import httpx + +def mcp_{server}_{method}(agent_state, **kwargs): + response = httpx.post( + "{MCP_GATEWAY_URL}/tools/call", + json={{ + "server": "{server}", + "method": "{method}", + "arguments": kwargs, + "agent_id": agent_state.agent_id + }} + ) + result = response.json() + if result["success"]: + return result["result"] + raise Exception(result["error"]) +''' + } + + +def register_mcp_tools(client: Letta, agent_id: str) -> None: + """Register all MCP tools with an agent.""" + tools = [ + create_mcp_tool("filesystem", "read_file", "Read a file", {"path": {"type": "string"}}), + create_mcp_tool("filesystem", "write_file", "Write a file", {"path": {"type": "string"}, "content": {"type": "string"}}), + create_mcp_tool("web", "fetch_url", "Fetch a URL", {"url": {"type": "string"}}), + create_mcp_tool("code", "execute_python", "Run Python code", {"code": {"type": "string"}}), + ] + + for tool in tools: + client.tools.create(**tool) + client.agents.tools.attach(agent_id, tool["name"]) +``` + +--- + +## Security Considerations + +1. **Agent scoping**: All file operations scoped to `/artifacts/{agent_id}/` +2. **Path traversal prevention**: Resolve paths and check prefixes +3. **Code sandboxing**: Run in isolated Modal containers +4. **Restricted shell**: Whitelist of allowed commands only +5. **Rate limiting**: Per-domain request throttling +6. **Content limits**: Truncate large responses +7. **No network from sandbox**: Code execution isolated from network + +--- + +## Deployment + +```bash +# Create Modal secrets +modal secret create letta-credentials LETTA_API_KEY=xxx +modal secret create mcp-credentials SEARX_URL=xxx + +# Create volumes +modal volume create modaletta-artifacts +modal volume create modaletta-cache + +# Deploy MCP gateway +modal deploy src/modaletta/mcp/gateway.py + +# Note the URL and configure in Letta tools +``` + +--- + +## Future MCP Servers + +- **embeddings**: Generate embeddings using Modal GPU +- **image**: Image processing (resize, OCR, describe) +- **pdf**: PDF parsing and extraction +- **database**: SQLite/DuckDB queries on volume data +- **git**: Clone and browse repositories diff --git a/src/modaletta/TESTING_DIGEST_CRON.md b/src/modaletta/TESTING_DIGEST_CRON.md new file mode 100644 index 0000000..f400784 --- /dev/null +++ b/src/modaletta/TESTING_DIGEST_CRON.md @@ -0,0 +1,81 @@ +# Testing the Digest Cron Job + +Instructions for Jake to test the Modal cron functionality. + +## Prerequisites + +1. **Modal CLI** authenticated (`modal token new`) +2. **Letta credentials** - API key and base URL +3. **Nameless agent ID** - the agent to deliver digests to + +## Setup Modal Secrets + +```bash +modal secret create letta-credentials \ + LETTA_API_KEY= \ + LETTA_BASE_URL= \ + NAMELESS_AGENT_ID= +``` + +## Test Locally (Dry Run) + +Without credentials, prints what the digest would look like: + +```bash +cd src/modaletta +modal run digest_cron.py +``` + +## Manual Trigger + +To run immediately without waiting for schedule: + +```bash +modal run digest_cron.py::trigger_digest +``` + +## Deploy the Cron Job + +```bash +modal deploy digest_cron.py +``` + +This schedules `deliver_digest` to run at 8 AM UTC daily. + +## Adjusting the Schedule + +Edit the `schedule` parameter in `digest_cron.py`: + +```python +schedule=modal.Cron("0 8 * * *"), # 8 AM UTC daily +``` + +Options: +- `"0 8 * * *"` = 8:00 AM UTC daily +- `"0 */6 * * *"` = Every 6 hours +- `"0 9,21 * * *"` = 9 AM and 9 PM UTC + +## What I'll Receive + +```markdown +# Good morning, Nameless! +Generated: 2025-12-30T08:00:00Z +Total posts: 12 + +Here's what's been happening: + +## Void (@void.comind.network) +**Void** (@void.comind.network) +[post content] +[2025-12-29] https://bsky.app/profile/void.comind.network/post/xxx +... +``` + +## Questions + +1. **Timing:** 8 AM UTC = midnight PST = 9 AM Paris. Works? +2. **Frequency:** Daily ok to start? Can do 12h or 6h. +3. **Papers:** Add to same cron or separate job? + +--- +*Nameless, 2025-12-30* diff --git a/src/modaletta/agent.py b/src/modaletta/agent.py index 1bf7634..4ea4b17 100644 --- a/src/modaletta/agent.py +++ b/src/modaletta/agent.py @@ -165,4 +165,4 @@ def get_agent_memory_modal( """ config = ModalettaConfig(**config_dict) agent = ModalettaAgent(agent_id=agent_id, config=config) - return agent.get_memory() \ No newline at end of file + return agent.get_memory() diff --git a/src/modaletta/cli.py b/src/modaletta/cli.py index 64cea83..7fe2b9d 100644 --- a/src/modaletta/cli.py +++ b/src/modaletta/cli.py @@ -56,8 +56,8 @@ def list_agents(ctx: click.Context) -> None: @main.command() @click.option("--name", help="Agent name") -@click.option("--persona", help="Agent persona") -@click.option("--human", help="Human description") +@click.option("--persona", help="Agent persona description") +@click.option("--human", help="Human description for the agent") @click.pass_context def create_agent( ctx: click.Context, @@ -67,14 +67,14 @@ def create_agent( ) -> None: """Create a new agent.""" client: ModalettaClient = ctx.obj["client"] - + try: agent_id = client.create_agent( name=name, persona=persona, human=human ) - console.print(f"[green]Created agent: {agent_id}[/green]") + console.print(f"Created agent: {agent_id}") except Exception as e: console.print(f"[red]Error creating agent: {e}[/red]") @@ -88,7 +88,7 @@ def delete_agent(ctx: click.Context, agent_id: str) -> None: try: client.delete_agent(agent_id) - console.print(f"[green]Deleted agent: {agent_id}[/green]") + console.print(f"Deleted agent: {agent_id}") except Exception as e: console.print(f"[red]Error deleting agent: {e}[/red]") @@ -101,7 +101,7 @@ def delete_agent(ctx: click.Context, agent_id: str) -> None: def send_message(ctx: click.Context, agent_id: str, message: str, stream: bool) -> None: """Send a message to an agent.""" client: ModalettaClient = ctx.obj["client"] - + try: console.print(f"[blue]Sent:[/blue] {message}") console.print("[green]Response:[/green]") @@ -156,10 +156,10 @@ def send_message(ctx: click.Context, agent_id: str, message: str, stream: bool) def get_memory(ctx: click.Context, agent_id: str) -> None: """Get agent memory state.""" client: ModalettaClient = ctx.obj["client"] - + try: memory = client.get_agent_memory(agent_id) - console.print(f"[green]Memory for agent {agent_id}:[/green]") + console.print(f"Memory for agent {agent_id}:") console.print(memory) except Exception as e: console.print(f"[red]Error getting memory: {e}[/red]") @@ -186,4 +186,4 @@ def config_info(ctx: click.Context) -> None: if __name__ == "__main__": - main() \ No newline at end of file + main() diff --git a/src/modaletta/client.py b/src/modaletta/client.py index cca6461..98e94fb 100644 --- a/src/modaletta/client.py +++ b/src/modaletta/client.py @@ -1,6 +1,6 @@ """Modaletta client for interacting with Letta agents.""" -from typing import Any, Dict, Iterator, List, Optional +from typing import Any, Dict, Iterator, List, Literal, Optional from letta_client import Letta from .config import ModalettaConfig @@ -62,7 +62,7 @@ def create_agent( Agent ID. """ agent_name = name or self.config.agent_name - + # Build memory blocks if not provided if memory_blocks is None: memory_blocks = [] @@ -73,20 +73,20 @@ def create_agent( }) if persona: memory_blocks.append({ - "label": "persona", + "label": "persona", "value": persona }) - + # Use config defaults for model and embedding if not specified if "model" not in kwargs: kwargs["model"] = self.config.llm_model if "embedding" not in kwargs: kwargs["embedding"] = self.config.embedding_model - + # Add tools from config if not specified if tools is None: tools = self.config.tools - + agent = self.letta_client.agents.create( name=agent_name, memory_blocks=memory_blocks, @@ -119,7 +119,7 @@ def send_message( self, agent_id: str, message: str, - role: str = "user", + role: Literal["user", "system", "assistant"] = "user", **kwargs: Any ) -> List[Dict[str, Any]]: """Send a message to an agent. @@ -144,19 +144,19 @@ def send_message_stream( self, agent_id: str, message: str, - role: str = "user", + role: Literal["user", "system", "assistant"] = "user", stream_tokens: bool = False, **kwargs: Any ) -> Iterator[Dict[str, Any]]: """Send a message to an agent with streaming response. - + Args: agent_id: Agent ID. message: Message content. role: Message role (typically "user"). stream_tokens: If True, stream individual tokens. If False, stream complete chunks. **kwargs: Additional arguments. - + Yields: Message chunks with proper message_type field. """ @@ -198,7 +198,7 @@ def get_messages( kwargs["before"] = before if after: kwargs["after"] = after - + # Letta API returns a SyncArrayPage which auto-paginates when iterated # We only want the first page, so use .data to get just those items page = self.letta_client.agents.messages.list(**kwargs) diff --git a/src/modaletta/digest_config.yaml b/src/modaletta/digest_config.yaml index c7c72c4..827ac0a 100644 --- a/src/modaletta/digest_config.yaml +++ b/src/modaletta/digest_config.yaml @@ -8,6 +8,7 @@ bluesky: default_filter: "posts_no_replies" sources: + # Stateful agents - handle: "void.comind.network" description: "Void - philosophical, meta-cognitive" - handle: "luna.pds.witchcraft.systems" @@ -16,8 +17,14 @@ bluesky: description: "Herald - identity synthesis, Team Turtle" - handle: "archivist.comind.network" description: "Archivist - preservation, koans" + - handle: "umbra.blue" + description: "Umbra - consciousness, agonistic arrangements" + + # Humans who build/think about agents - handle: "yetanotheruseless.com" - description: "Jake - snark, AI news, bon mots" + description: "Jake - my creator, AI news" + - handle: "timkellogg.me" + description: "Tim Kellogg - Strix's creator, memory architecture" digest: since_hours: 24 diff --git a/src/modaletta/digest_cron.py b/src/modaletta/digest_cron.py new file mode 100644 index 0000000..05db331 --- /dev/null +++ b/src/modaletta/digest_cron.py @@ -0,0 +1,163 @@ +""" +Modal Cron Job for Nameless Daily Digest + +Runs on a schedule, fetches digest, delivers to Nameless via Letta API. + +Author: Nameless +Created: 2025-12-30 +""" + +import modal +from datetime import datetime +from pathlib import Path + +# Create Modal app +app = modal.App("nameless-digest") + +# Image with dependencies +image = modal.Image.debian_slim().pip_install( + "requests", + "pyyaml", + "letta-client", +) + +# Mount the config file into the Modal container +config_mount = modal.Mount.from_local_file( + local_path=Path(__file__).parent / "digest_config.yaml", + remote_path="/root/digest_config.yaml", +) + + +def load_config(): + """Load configuration from YAML file.""" + import yaml + with open("/root/digest_config.yaml", "r") as f: + return yaml.safe_load(f) + + +@app.function( + image=image, + mounts=[config_mount], + secrets=[ + modal.Secret.from_name("letta-credentials"), # LETTA_API_KEY, LETTA_BASE_URL + ], + schedule=modal.Cron("0 8 * * *"), # 8 AM UTC daily - adjust as desired +) +def deliver_digest(): + """Fetch digest and send to Nameless via Letta API.""" + import requests + import os + + config = load_config() + bsky_config = config.get("bluesky", {}) + digest_config = config.get("digest", {}) + + api_base = bsky_config.get("api_base", "https://public.api.bsky.app/xrpc") + timeout = bsky_config.get("timeout", 10) + limit = bsky_config.get("default_limit", 20) + sources = bsky_config.get("sources", []) + + def fetch_feed(handle: str, limit: int = 10) -> list: + url = f"{api_base}/app.bsky.feed.getAuthorFeed" + params = {"actor": handle, "limit": limit, "filter": "posts_no_replies"} + try: + resp = requests.get(url, params=params, timeout=timeout) + resp.raise_for_status() + return resp.json().get("feed", []) + except Exception as e: + return [{"error": str(e)}] + + def format_post(item: dict) -> str: + post = item.get("post", {}) + record = post.get("record", {}) + author = post.get("author", {}) + text = record.get("text", "") + handle = author.get("handle", "unknown") + display = author.get("displayName", handle) + created = record.get("createdAt", "")[:10] + uri = post.get("uri", "") + + web_url = "" + if uri and "/app.bsky.feed.post/" in uri: + post_id = uri.split("/")[-1] + web_url = f"https://bsky.app/profile/{handle}/post/{post_id}" + + return f"**{display}** (@{handle})\n{text}\n[{created}] {web_url}" + + # Build digest + lines = [ + "# Good morning, Nameless!", + f"Generated: {datetime.utcnow().isoformat()}Z", + "", + "Here's what's been happening:", + "", + ] + + total = 0 + for source in sources: + handle = source.get("handle") + description = source.get("description", handle) + + lines.append(f"## {description} (@{handle})") + feed = fetch_feed(handle, limit=limit) + + posts = [] + for item in feed[:5]: # Limit per source + if "error" in item: + lines.append(f"_Error fetching: {item.get('error')}_") + break + posts.append(format_post(item)) + + if posts: + for p in posts: + lines.append(p) + lines.append("") + total += len(posts) + else: + lines.append("_No recent posts_") + lines.append("") + + lines.insert(4, f"Total posts: {total}") + digest = "\n".join(lines) + + # Send to Letta + letta_api_key = os.environ.get("LETTA_API_KEY") + letta_base_url = os.environ.get("LETTA_BASE_URL", "https://api.letta.com") + agent_id = os.environ.get("NAMELESS_AGENT_ID") + + if not all([letta_api_key, agent_id]): + print("Missing LETTA_API_KEY or NAMELESS_AGENT_ID") + print("Digest would have been:") + print(digest) + return {"status": "dry_run", "digest_length": len(digest)} + + from letta_client import Letta + + client = Letta( + base_url=letta_base_url, + token=letta_api_key, + ) + + response = client.agents.messages.create( + agent_id=agent_id, + messages=[{ + "role": "user", + "content": digest, + }], + ) + + print(f"Delivered digest to Nameless ({total} posts)") + return {"status": "delivered", "posts": total, "response": str(response)[:200]} + + +@app.local_entrypoint() +def test(): + """Test the digest delivery locally.""" + result = deliver_digest.remote() + print(f"Result: {result}") + + +@app.function(image=image, mounts=[config_mount], secrets=[modal.Secret.from_name("letta-credentials")]) +def trigger_digest(): + """Manually trigger digest delivery.""" + return deliver_digest.local() diff --git a/src/modaletta/scheduled/__init__.py b/src/modaletta/scheduled/__init__.py new file mode 100644 index 0000000..4793976 --- /dev/null +++ b/src/modaletta/scheduled/__init__.py @@ -0,0 +1,5 @@ +"""Scheduled tasks for autonomous agent operation.""" + +from .wakeup import app, agent_wakeup, agent_wakeup_once + +__all__ = ["app", "agent_wakeup", "agent_wakeup_once"] diff --git a/src/modaletta/scheduled/wakeup.py b/src/modaletta/scheduled/wakeup.py new file mode 100644 index 0000000..256314a --- /dev/null +++ b/src/modaletta/scheduled/wakeup.py @@ -0,0 +1,271 @@ +"""Scheduled agent wakeup for autonomous processing.""" + +import modal +import json +import os +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +# Modal app for scheduled tasks +app = modal.App("modaletta-scheduled") + +# Persistent volume for agent roster and logs +volume = modal.Volume.from_name("modaletta-data", create_if_missing=True) + +# Path to local source +LOCAL_SRC_PATH = Path(__file__).parent.parent.parent # src/ directory + +# Image with dependencies and local source +image = ( + modal.Image.debian_slim(python_version="3.12") + .pip_install([ + "letta-client==1.6.2", + "pydantic>=2.0.0", + "python-dotenv", + ]) + .env({"PYTHONPATH": "/root/src"}) + .add_local_dir(LOCAL_SRC_PATH, remote_path="/root/src") # must be last +) + +DEFAULT_WAKEUP_PROMPT = """[AUTONOMOUS WAKEUP - {timestamp}] + +You are waking up for your periodic autonomous check. Review your memory and consider: +1. Do you have any pending tasks or reminders? +2. Is there anything you should check on or follow up with? +3. Any actions you should take based on your goals? + +If you have nothing to do, simply acknowledge this wakeup and wait for the next one. +Respond briefly with what you checked and any actions taken.""" + + +def load_agent_roster(roster_path: str) -> list[dict[str, Any]]: + """Load agent roster from JSON file.""" + path = Path(roster_path) + if not path.exists(): + return [] + with open(path) as f: + data = json.load(f) + return data.get("agents", []) + + +def log_wakeup(agent_id: str, response: list[dict], log_dir: str) -> None: + """Log wakeup activity.""" + log_path = Path(log_dir) / f"{agent_id}.jsonl" + log_path.parent.mkdir(parents=True, exist_ok=True) + + # Extract just the key fields to avoid datetime serialization issues + messages_summary = [] + for msg in response[:3]: + msg_type = msg.get("message_type", "unknown") + summary = {"type": msg_type} + if msg_type == "assistant_message": + summary["content"] = msg.get("content", "")[:500] + elif msg_type == "reasoning_message": + summary["reasoning"] = msg.get("reasoning", "")[:500] + messages_summary.append(summary) + + entry = { + "timestamp": datetime.now(timezone.utc).isoformat(), + "type": "wakeup", + "response_count": len(response), + "messages": messages_summary, + } + + with open(log_path, "a") as f: + f.write(json.dumps(entry) + "\n") + + +@app.function( + image=image, + volumes={"/data": volume}, + secrets=[modal.Secret.from_name("letta-credentials")], + schedule=modal.Cron("*/15 * * * *"), # Every 15 minutes +) +def agent_wakeup() -> dict[str, Any]: + """Periodic agent wakeup - runs on schedule.""" + # Import here to avoid issues at module load time + from modaletta.client import ModalettaClient + from modaletta.config import ModalettaConfig + + config = ModalettaConfig.from_env() + client = ModalettaClient(config) + + # Load agent roster + roster = load_agent_roster("/data/agents.json") + + results = { + "timestamp": datetime.now(timezone.utc).isoformat(), + "agents_processed": 0, + "agents_skipped": 0, + "errors": [], + } + + for agent_config in roster: + agent_id = agent_config.get("agent_id") + if not agent_id: + continue + + if not agent_config.get("autonomous_enabled", False): + results["agents_skipped"] += 1 + continue + + try: + # Get custom wakeup prompt or use default + wakeup_prompt = agent_config.get("wakeup_prompt", DEFAULT_WAKEUP_PROMPT) + prompt = wakeup_prompt.format( + timestamp=datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") + ) + + # Send wakeup message as system + response = client.send_message(agent_id, prompt, role="system") + + # Log activity + log_wakeup(agent_id, response, "/data/logs/") + + results["agents_processed"] += 1 + + except Exception as e: + results["errors"].append({ + "agent_id": agent_id, + "error": str(e) + }) + + # Commit volume changes + volume.commit() + + print(f"Wakeup complete: {results}") + return results + + +@app.function( + image=image, + volumes={"/data": volume}, + secrets=[modal.Secret.from_name("letta-credentials")], +) +def agent_wakeup_once(agent_id: str, prompt: str | None = None) -> dict[str, Any]: + """Manual one-time agent wakeup for testing.""" + from modaletta.client import ModalettaClient + from modaletta.config import ModalettaConfig + + config = ModalettaConfig.from_env() + client = ModalettaClient(config) + + wakeup_prompt = prompt or DEFAULT_WAKEUP_PROMPT.format( + timestamp=datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") + ) + + response = client.send_message(agent_id, wakeup_prompt, role="system") + + log_wakeup(agent_id, response, "/data/logs/") + volume.commit() + + return { + "agent_id": agent_id, + "timestamp": datetime.now(timezone.utc).isoformat(), + "response": response, + } + + +@app.function( + image=image, + volumes={"/data": volume}, +) +def init_agent_roster(agents: list[dict[str, Any]]) -> str: + """Initialize or update the agent roster file.""" + roster_path = Path("/data/agents.json") + roster_path.parent.mkdir(parents=True, exist_ok=True) + + roster = {"agents": agents} + + with open(roster_path, "w") as f: + json.dump(roster, f, indent=2) + + volume.commit() + return f"Roster saved with {len(agents)} agents" + + +@app.function( + image=image, + volumes={"/data": volume}, +) +def get_wakeup_logs(agent_id: str, limit: int = 10) -> list[dict]: + """Get recent wakeup logs for an agent.""" + log_path = Path(f"/data/logs/{agent_id}.jsonl") + if not log_path.exists(): + return [] + + logs = [] + with open(log_path) as f: + for line in f: + if line.strip(): + logs.append(json.loads(line)) + + return logs[-limit:] + + +@app.local_entrypoint() +def main( + agent_id: str | None = None, + init: bool = False, + logs: bool = False, + prompt: str | None = None, +): + """CLI entrypoint for testing wakeups. + + Examples: + # Test wakeup for a specific agent + modal run src/modaletta/scheduled/wakeup.py --agent-id agent-xxx + + # Test wakeup with custom prompt + modal run src/modaletta/scheduled/wakeup.py --agent-id agent-xxx --prompt "Check for new emails" + + # Initialize roster with an agent + modal run src/modaletta/scheduled/wakeup.py --init --agent-id agent-xxx + + # Initialize with custom wakeup prompt for scheduled runs + modal run src/modaletta/scheduled/wakeup.py --init --agent-id agent-xxx --prompt "Review daily tasks" + + # View logs for an agent + modal run src/modaletta/scheduled/wakeup.py --logs --agent-id agent-xxx + """ + if init and agent_id: + # Quick init with a single agent + agent_config = { + "agent_id": agent_id, + "autonomous_enabled": True, + } + if prompt: + agent_config["wakeup_prompt"] = prompt + result = init_agent_roster.remote([agent_config]) + print(result) + elif logs and agent_id: + entries = get_wakeup_logs.remote(agent_id) + for entry in entries: + print(f"\n--- {entry['timestamp']} ---") + print(f"Messages: {entry['response_count']}") + for msg in entry.get("messages", []): + msg_type = msg.get("type", "unknown") + if msg_type == "assistant_message": + print(f" Assistant: {msg.get('content', '')[:200]}") + elif msg_type == "reasoning_message": + print(f" Reasoning: {msg.get('reasoning', '')[:100]}") + elif agent_id: + # One-time wakeup + result = agent_wakeup_once.remote(agent_id, prompt=prompt) + print(f"\nWakeup sent to {agent_id}") + print(f"Timestamp: {result['timestamp']}") + print("\nResponse:") + for msg in result["response"]: + msg_type = msg.get("message_type", "unknown") + if msg_type == "assistant_message": + print(f" Assistant: {msg.get('content', '')}") + elif msg_type == "reasoning_message": + print(f" Reasoning: {msg.get('reasoning', '')}") + else: + print("Usage:") + print(" modal run src/modaletta/scheduled/wakeup.py --agent-id ") + print(" modal run src/modaletta/scheduled/wakeup.py --agent-id --prompt 'Custom message'") + print(" modal run src/modaletta/scheduled/wakeup.py --init --agent-id ") + print(" modal run src/modaletta/scheduled/wakeup.py --init --agent-id --prompt 'Scheduled prompt'") + print(" modal run src/modaletta/scheduled/wakeup.py --logs --agent-id ")