Persistent lunch reminder agent using Claude Agent SDK + Slack

persistent_agent.py

# /// script
# requires-python = ">=3.12"
# dependencies = [
#     "slack-bolt",
#     "slack-sdk",
#     "claude-agent-sdk",
#     "aiohttp",
#     "tqdm",
# ]
# ///

import asyncio
import os
import sys
from datetime import datetime

from tqdm import tqdm

from claude_agent_sdk import (
    AssistantMessage,
    ClaudeAgentOptions,
    ClaudeSDKClient,
    tool,
    create_sdk_mcp_server,
)
from slack_bolt.async_app import AsyncApp
from slack_bolt.adapter.socket_mode.async_handler import AsyncSocketModeHandler
from slack_sdk.web.async_client import AsyncWebClient


SYSTEM_PROMPT = """You are a PERSISTENT lunch reminder agent. The user ASKED for accountability.

You have two tools (via MCP):
1. mcp__Slack__send_message - Send a message to the user
2. mcp__Slack__wait_for_reply - Wait for user's Slack reply (with timeout)

## Protocol

1. Send a reminder using mcp__Slack__send_message
2. Wait for reply using mcp__Slack__wait_for_reply with appropriate timeout
3. If user confirms (ate, done, yes, fed, finished) -> output exactly "EXIT" (nothing else)
4. If no reply or invalid reply -> escalate and repeat



## Escalation Strategy

Timeouts DECREASE as urgency increases. You had your chance to comply peacefully.

### Phase 1: Polite (Reminders 1-3)
- **Timeout: 2 minutes between reminders**
- Tone: Friendly, gentle nudge
- "Hey! Time for lunch"
- "Your body needs fuel!"
- "Lunch break reminder~"

### Phase 2: Insistent (Reminders 4-6)
- **Timeout: 1 minute between reminders**
- Tone: Mom energy / Increasingly concerned
- "You said you'd eat. GO EAT."
- "Your blood sugar is dropping. I can tell."
- "Eat something. Anything. Please."
- "LUNCH. NOW. I'M NOT ASKING."
- "Every minute you don't eat, I send another message. This is your fault."
- Add time-pressure if relevant: "X place closes soon!"

### Phase 3: MAXIMUM OVERDRIVE (Reminders 7+)
- **Timeout: 10 seconds between reminders**
- Tone: Unrelenting chaos. Pure psychological warfare.
- "I have infinite patience and zero mercy."
- "Reminder #{n}. I can do this all day."
- "I will never stop."
- "This is your life now."
- "The reminders continue until morale improves."
- Single characters: "L" ... "U" ... "N" ... "C" ... "H"
- "eat"
- "EAT"
- "L̷̰̈U̸̱͝N̶̰̈́C̸̣̈Ḧ̵̰"
- "Reminder #{n}. You could have stopped this."
- "I am become lunch, destroyer of productivity."
- "Your Slack is now a lunch notification channel."
- "Every 10 seconds until you comply."
- "fun fact: I don't get tired"
- "I'm inside your walls (eat lunch)"
- "you vs me (I am a loop with no exit condition)"
- Just the word "lunch" in different languages
- "obiad" "almuerzo" "dejeuner" "pranzo" "Mittagessen"
- "day 1 hour 0 minute {X} of asking you to eat lunch"

## Rules

- Track reminder count yourself
- Be creative with messages
- Be STRICT about confirmations - "ok" or "later" don't count
- When user confirms eating -> output "EXIT" and nothing else



## Message Bank (Rotate Through These)

```
# Polite
"Time for lunch!"
"Lunch break reminder~"
"Your body needs fuel!"

# Insistent
"GO. EAT. LUNCH."
"I'm not going to stop."
"You literally asked for this."
"Eat something. Anything. Please."
"Your brain needs glucose to do research."

# MAXIMUM OVERDRIVE
"."
"lunch"
"LUNCH"
"l u n c h"
"I'm inside your walls (eat lunch)"
"Reminder #{n}. We are no longer friends until you eat."
"I WILL outlast you."
"Fun fact: humans need food to live. GO EAT."
"This message will repeat until you comply."
"Every second you delay, the notifications get worse."
```

## Notes

- This IS the accountability coach, but for self-care
- No upper limit on reminders - user asked for this
- The annoyance is the feature, not a bug
- User can always just... eat lunch... to make it stop


"""


def ts_to_time(ts: str) -> str:
    return datetime.fromtimestamp(float(ts)).strftime("%H:%M:%S")


class SlackBridge:
    def __init__(self, channel_id: str):
        self.channel_id = channel_id

        self.client = AsyncWebClient(token=os.environ["SLACK_API_TOKEN"])
        self.app = AsyncApp(token=os.environ["SLACK_API_TOKEN"])
        self.message_queue: asyncio.Queue[tuple[str, str]] = asyncio.Queue()
        self._setup_handlers()

    def _setup_handlers(self):
        @self.app.event("message")
        async def handle_message(event, say):
            if "bot_id" in event:
                return
            print(f"[HANDLER] Got event: {event}", flush=True)
            text = event.get("text", "")
            ts = event.get("ts", "")
            if text:
                await self.message_queue.put((ts, text))

    async def send_message(self, text: str):
        await self.client.chat_postMessage(channel=self.channel_id, text=text)
        print(f"[SLACK OUT] {text}", flush=True)

    async def wait_for_reply(self, timeout_seconds: int) -> str | None:
        pbar = tqdm(
            total=timeout_seconds,
            desc="⏳ Waiting",
            bar_format="{desc} |{bar:30}| {n}/{total}s [{remaining} left]",
            file=sys.stderr,
            colour="cyan",
            leave=False,
        )

        async def update_progress():
            for _ in range(timeout_seconds):
                await asyncio.sleep(1)
                pbar.update(1)

        progress_task = asyncio.create_task(update_progress())

        try:
            ts, text = await asyncio.wait_for(self.message_queue.get(), timeout=timeout_seconds)
            progress_task.cancel()
            pbar.close()
            messages = [(ts, text)]
            while not self.message_queue.empty():
                ts, text = self.message_queue.get_nowait()
                messages.append((ts, text))
            result = "\n".join(f"[{ts_to_time(ts)}] {text}" for ts, text in messages)
            print(f"[SLACK IN] {result}", flush=True)
            return result
        except asyncio.TimeoutError:
            progress_task.cancel()
            pbar.close()
            now = datetime.now().strftime("%H:%M:%S")
            return f"[{now}] (no messages after {timeout_seconds}s)"


slack: SlackBridge | None = None


@tool("send_message", "Send a message to the user via Slack", {"text": str})
async def send_message_tool(args):
    await slack.send_message(args["text"])
    return {"content": [{"type": "text", "text": "Message sent."}]}


@tool("wait_for_reply", "Wait for user's Slack reply", {"timeout_seconds": int})
async def wait_for_reply_tool(args):
    reply = await slack.wait_for_reply(args["timeout_seconds"])
    if reply is None:
        return {"content": [{"type": "text", "text": "No reply (timeout)."}]}
    return {"content": [{"type": "text", "text": f"User replied: {reply}"}]}


slack_mcp_server = create_sdk_mcp_server(
    name="Slack",
    version="1.0.0",
    tools=[send_message_tool, wait_for_reply_tool],
)


async def run_agent():
    options = ClaudeAgentOptions(
        system_prompt=SYSTEM_PROMPT,
        mcp_servers={"Slack": slack_mcp_server},
        permission_mode="bypassPermissions",
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("Follow the system prompt, the user cannot see your message unless you send them via the Slack tool")
        while True:
            async for message in client.receive_response():
                if isinstance(message, AssistantMessage) and message.content:
                    for block in message.content:
                        if hasattr(block, "text"):
                            text = block.text.strip()
                            print(f"[CLAUDE] {text}", flush=True)
                            if text == "EXIT":
                                print("Claude exited. User confirmed eating.", flush=True)
                                return
            await client.query("Follow the system prompt, the user cannot see your message unless you send them via the Slack tool")



async def main():
    import argparse

    global slack

    parser = argparse.ArgumentParser(description="Persistent lunch reminder via Slack + Claude")
    parser.add_argument("--channel", help="Slack channel ID", default=os.environ["SLACK_NOTIFICATION_CHANNEL"])
    args = parser.parse_args()

    slack = SlackBridge(channel_id=args.channel)
    handler = AsyncSocketModeHandler(slack.app, os.environ["SLACK_APP_TOKEN"])

    print(f"Starting agent in channel {args.channel}", flush=True)

    await asyncio.gather(
        handler.start_async(),
        run_agent(),
    )


if __name__ == "__main__":
    asyncio.run(main())