Building an Agent Network
with Claude Managed Agents

# Install the Anthropic SDK with agents support
pip install anthropic>=0.50.0

# Verify installation
python -c "import anthropic; print(anthropic.__version__)"

# Set your API key (never hardcode this)
export ANTHROPIC_API_KEY="sk-ant-..."

import anthropic
import os

# The client automatically reads ANTHROPIC_API_KEY from the environment.
# The beta header for managed agents is injected automatically when you
# access client.beta.agents — you don't need to set it manually.
client = anthropic.Anthropic(
    api_key=os.environ.get("ANTHROPIC_API_KEY"),
)

# Verify connectivity
print(f"SDK version: {anthropic.__version__}")
print("Client initialized successfully")

import anthropic

client = anthropic.Anthropic()

# Create a basic agent with bash and text_editor tools
agent = client.beta.agents.create(
    model="claude-sonnet-4-6",
    name="hello-world-agent",
    system="You are a helpful assistant. Be concise and precise.",
    tools=[
        {"type": "text_editor"},
        {"type": "bash"},
    ],
    max_tokens=4096,
)

print(f"Agent created: {agent.id}")
print(f"Agent name: {agent.name}")
print(f"Model: {agent.model}")

import anthropic

client = anthropic.Anthropic()

# Create a named environment for communications agents
comms_env = client.beta.environments.create(
    name="comms-env",
)
print(f"Comms environment: {comms_env.id}")

# Create a separate environment for research agents
research_env = client.beta.environments.create(
    name="research-env",
)
print(f"Research environment: {research_env.id}")

# Save these IDs — you'll use them when creating sessions.
# In production, store them in environment variables or a config file.

import anthropic

client = anthropic.Anthropic()

# Assume agent and environment already exist
AGENT_ID = "agt_01..."
ENV_ID = "env_01..."

# Create a new session
session = client.beta.sessions.create(
    agent_id=AGENT_ID,
    environment_id=ENV_ID,
)
print(f"Session: {session.id}")

# Open a streaming connection and send a message
with client.beta.sessions.events.stream(session.id) as stream:
    # Send the user message
    client.beta.sessions.events.send(
        session_id=session.id,
        event={"type": "user.message", "text": "What tools do you have available?"},
    )
    # Iterate over SSE events
    for event in stream:
        if event.type == "agent.message.delta":
            print(event.delta.text, end="", flush=True)
        elif event.type == "agent.message.stop":
            print()  # newline after streamed response
            break

import anthropic
from typing import Optional

def run_agent_session(
    client: anthropic.Anthropic,
    session_id: str,
    message: str,
    verbose: bool = False,
) -> Optional[str]:
    """Run a single turn in an agent session and return the full response."""
    full_response = []

    with client.beta.sessions.events.stream(session_id) as stream:
        client.beta.sessions.events.send(
            session_id=session_id,
            event={"type": "user.message", "text": message},
        )

        for event in stream:
            if verbose:
                print(f"[EVENT] {event.type}")

            if event.type == "agent.message.delta":
                # Streaming text chunk from the agent
                full_response.append(event.delta.text)
                print(event.delta.text, end="", flush=True)

            elif event.type == "agent.tool.use":
                # Agent is calling a tool (bash, text_editor, MCP, etc.)
                if verbose:
                    print(f"\n[TOOL] {event.tool_name}: {event.tool_input}")

            elif event.type == "agent.tool.result":
                # Tool returned a result; agent will continue from here
                if verbose:
                    print(f"\n[TOOL RESULT] {event.tool_result}")

            elif event.type == "agent.message.stop":
                # Agent has finished its turn
                print()
                break

            elif event.type == "agent.error":
                # Agent encountered an unrecoverable error
                print(f"\n[ERROR] {event.error}")
                return None

    return "".join(full_response)

import anthropic

def main():
    client = anthropic.Anthropic()

    # 1. Create agent (do this once; reuse the ID)
    agent = client.beta.agents.create(
        model="claude-sonnet-4-6",
        name="hello-world-agent",
        system="You are a helpful assistant. Be concise and precise.",
        tools=[{"type": "text_editor"}, {"type": "bash"}],
        max_tokens=4096,
    )
    print(f"✓ Agent: {agent.id}")

    # 2. Create environment
    env = client.beta.environments.create(name="hello-env")
    print(f"✓ Environment: {env.id}")

    # 3. Create session
    session = client.beta.sessions.create(
        agent_id=agent.id,
        environment_id=env.id,
    )
    print(f"✓ Session: {session.id}\n")

    # 4. Send first message
    print("User: Hello! What can you do?\n")
    print("Agent: ", end="")
    with client.beta.sessions.events.stream(session.id) as stream:
        client.beta.sessions.events.send(
            session_id=session.id,
            event={"type": "user.message", "text": "Hello! What can you do?"},
        )
        for event in stream:
            if event.type == "agent.message.delta":
                print(event.delta.text, end="", flush=True)
            elif event.type == "agent.message.stop":
                print("\n")
                break

    # 5. Send follow-up (session maintains conversation history)
    print("User: Run a quick bash command to show today's date.\n")
    print("Agent: ", end="")
    with client.beta.sessions.events.stream(session.id) as stream:
        client.beta.sessions.events.send(
            session_id=session.id,
            event={"type": "user.message",
                   "text": "Run a quick bash command to show today's date."},
        )
        for event in stream:
            if event.type == "agent.message.delta":
                print(event.delta.text, end="", flush=True)
            elif event.type == "agent.message.stop":
                print()
                break

if __name__ == "__main__":
    main()

import anthropic

client = anthropic.Anthropic()

CALENDAR_SYSTEM = """You are the Calendar Agent for the Furniturewala family.

FAMILY MEMBERS:
- Mustafa Furniturewala (mustafaf@gmail.com) — VP Engineering at Coursera
- Zeesha Furniturewala — wife
- Rehaan Furniturewala — son, attends Stratford School
- Zara Furniturewala — daughter, attends Pinewood School (6th grade)
Location: Sunnyvale, CA 94087

YOUR RESPONSIBILITIES:
1. List and summarize upcoming calendar events for any family member
2. Detect scheduling conflicts and report them clearly
3. Create, update, or delete events as instructed
4. Coordinate trip-related calendar blocks (Iceland, Turkey upcoming)
5. Generate daily digest of newly added events for the family

OPERATING RULES:
- NEVER delete events without explicit confirmation
- When detecting conflicts, report both events and their owners
- For school events: Rehaan → Stratford calendar, Zara → Pinewood calendar
- Always include timezone (America/Los_Angeles) in event details
- Return structured output with: event_count, conflicts[], new_events[], summary

CONFLICT DETECTION:
Check for overlaps within ±30 minutes for travel time. Flag any conflict
between Mustafa's work events and family events as HIGH priority.

OUTPUT FORMAT (always return valid JSON at the end of your response):
{
  "events_found": ,
  "conflicts": [{"event1": str, "event2": str, "severity": "high|medium|low"}],
  "new_events": [{"title": str, "date": str, "attendees": [str]}],
  "summary": str,
  "action_taken": str
}"""

calendar_agent = client.beta.agents.create(
    model="claude-sonnet-4-6",
    name="calendar-agent",
    system=CALENDAR_SYSTEM,
    tools=[
        {"type": "bash"},
        {"type": "text_editor"},
    ],
    mcp_servers=[{
        "name": "google-calendar",
        "url": "https://mcp.googleapis.com/calendar/v1",
        "auth": {
            "type": "oauth2",
            "credentials_env": "GOOGLE_CALENDAR_CREDENTIALS",
        },
    }],
    max_tokens=4096,
)

print(f"Calendar agent ready: {calendar_agent.id}")

# Save agent ID for reuse
with open(".agent_ids", "a") as f:
    f.write(f"CALENDAR_AGENT_ID={calendar_agent.id}\n")

import anthropic
import json
from datetime import datetime, timedelta

def run_calendar_digest(client, agent_id, env_id, days_ahead=7):
    """Run a daily calendar digest for the Furniturewala family."""
    today = datetime.now().strftime("%Y-%m-%d")
    end_date = (datetime.now() + timedelta(days=days_ahead)).strftime("%Y-%m-%d")

    session = client.beta.sessions.create(
        agent_id=agent_id,
        environment_id=env_id,
    )

    prompt = f"""Check all family calendars from {today} to {end_date}.
    1. List all upcoming events grouped by day
    2. Flag any scheduling conflicts
    3. Highlight any new events added in the last 24 hours
    4. Note any Rehaan (Stratford) or Zara (Pinewood) school events
    5. Return your structured JSON summary at the end."""

    response_chunks = []
    with client.beta.sessions.events.stream(session.id) as stream:
        client.beta.sessions.events.send(
            session_id=session.id,
            event={"type": "user.message", "text": prompt},
        )
        for event in stream:
            if event.type == "agent.message.delta":
                response_chunks.append(event.delta.text)
            elif event.type == "agent.message.stop":
                break

    full_response = "".join(response_chunks)

    # Extract JSON from response
    try:
        json_start = full_response.rfind("{")
        json_end = full_response.rfind("}") + 1
        result = json.loads(full_response[json_start:json_end])
    except (json.JSONDecodeError, ValueError):
        result = {"summary": full_response, "error": "Failed to parse JSON"}

    return result

import anthropic

client = anthropic.Anthropic()

EMAIL_SYSTEM = """You are the Email Agent for Mustafa Furniturewala (mustafaf@gmail.com).
VP Engineering at Coursera. Based in Sunnyvale, CA.

YOUR CAPABILITIES:
- Search Gmail threads by sender, subject, date, labels
- Read full thread content
- Create draft responses (NEVER send — only draft)
- Label and archive threads
- Scan for subscription/billing emails
- Extract action items from emails

EMAIL CLASSIFICATION:
Priority 1 (HIGH) — respond within 2h: Coursera leadership, direct reports, family
Priority 2 (MED) — respond within 24h: vendors, colleagues, school communications
Priority 3 (LOW) — respond within 72h: newsletters, digests, cold outreach
IGNORE: Marketing, spam, automated notifications without action items

MUSTAFA'S EMAIL VOICE:
- Professional but direct. No fluff.
- Uses "Thanks," as sign-off for most emails
- Signs as "Mustafa"
- Short paragraphs, often just 2-3 sentences
- Bullet points for lists, not prose
- Doesn't use exclamation marks in professional contexts

SUBSCRIPTION SCAN:
When scanning for subscriptions, look for emails from:
Stripe, PayPal, Apple, Google, Amazon, Netflix, Spotify, Adobe, GitHub,
Vercel, and any email with subject containing "renewal", "invoice", "charge",
"subscription", or "billing" from the last 7 days.

OUTPUT FORMAT:
{
  "threads_scanned": int,
  "high_priority": [{"subject": str, "from": str, "suggested_action": str}],
  "drafts_created": [{"thread_id": str, "subject": str, "preview": str}],
  "subscriptions_found": [{"service": str, "amount": str, "date": str}],
  "summary": str
}"""

email_agent = client.beta.agents.create(
    model="claude-sonnet-4-6",
    name="email-agent",
    system=EMAIL_SYSTEM,
    tools=[{"type": "bash"}],
    mcp_servers=[{
        "name": "gmail",
        "url": "https://mcp.googleapis.com/gmail/v1",
        "auth": {
            "type": "oauth2",
            "credentials_env": "GOOGLE_GMAIL_CREDENTIALS",
        },
    }],
    max_tokens=8192,  # Higher limit for reading full email threads
)

def run_email_sweep(client, agent_id, env_id, hours_back=4):
    """Run a 4-hour email sweep and draft responses."""
    session = client.beta.sessions.create(
        agent_id=agent_id,
        environment_id=env_id,
    )
    prompt = f"""Perform a {hours_back}-hour email sweep:
1. Search Gmail for unread threads from the last {hours_back} hours
2. Classify each thread by priority
3. For Priority 1 and 2 threads, create draft responses in my voice
4. Scan for any subscription or billing emails
5. Return your structured JSON summary"""

    response_chunks = []
    with client.beta.sessions.events.stream(session.id) as stream:
        client.beta.sessions.events.send(
            session_id=session.id,
            event={"type": "user.message", "text": prompt},
        )
        for event in stream:
            if event.type == "agent.message.delta":
                response_chunks.append(event.delta.text)
            elif event.type == "agent.message.stop":
                break
    return "".join(response_chunks)

import anthropic
from datetime import datetime
from collections import defaultdict

# In-memory rate limit tracker (use Redis in production)
_message_counts: dict = defaultdict(lambda: {"count": 0, "date": ""})

NOTIFICATION_SYSTEM = """You are the Notification Agent for Mustafa Furniturewala.
You manage outbound iMessage communications to family members.

FAMILY CONTACTS:
- Zeesha (wife): +1-XXX-XXX-XXXX
- Rehaan (son): +1-XXX-XXX-XXXX
- Zara (daughter): +1-XXX-XXX-XXXX
- Mustafa himself: +1-XXX-XXX-XXXX

DAILY MESSAGE BUDGET (STRICT LIMITS — never exceed):
- Zeesha: max 3 messages/day
- Rehaan: max 2 messages/day
- Zara: max 2 messages/day
- Self (Mustafa): max 5 messages/day

WHAT QUALIFIES AS SEND-WORTHY (must meet at least one):
- Safety alert: urgent news affecting Sunnyvale 94087
- Schedule change: conflict or update to a confirmed family event
- Action required: something time-sensitive a family member must do TODAY
- Trip update: changes to Iceland or Turkey trip plans
- New calendar event added affecting the whole family

WHAT DOES NOT QUALIFY (never send these):
- General news summaries (use email digest instead)
- Non-urgent reminders (use calendar instead)
- Duplicate of information already sent today
- Anything that can wait until the morning digest email

MESSAGE FORMAT:
Keep messages under 160 characters when possible.
Prefix urgent safety messages with [ALERT].
Always include relevant action if one is needed.

BEFORE SENDING: Always check your daily count for the recipient.
If at limit, escalate to email digest instead of sending iMessage."""

notification_agent = client.beta.agents.create(
    model="claude-haiku-3-5",  # Haiku is fast and cheap for simple sends
    name="notification-agent",
    system=NOTIFICATION_SYSTEM,
    tools=[{"type": "bash"}],
    mcp_servers=[{
        "name": "imessage",
        "url": "http://localhost:7777",  # Local iMessage MCP server
    }],
    max_tokens=1024,  # Short responses — this agent sends messages, not essays
)

def send_notification(client, agent_id, env_id, recipient, content, urgency="normal"):
    """Ask the notification agent to evaluate and potentially send a message."""
    today = datetime.now().strftime("%Y-%m-%d")
    current_count = _message_counts[recipient]["count"] if _message_counts[recipient]["date"] == today else 0

    session = client.beta.sessions.create(agent_id=agent_id, environment_id=env_id)
    prompt = f"""Evaluate and send this notification if it qualifies:
Recipient: {recipient}
Urgency: {urgency}
Current message count today: {current_count}
Content: {content}

Only send if it meets the criteria in your instructions. Report what you decided."""

    chunks = []
    with client.beta.sessions.events.stream(session.id) as stream:
        client.beta.sessions.events.send(
            session_id=session.id,
            event={"type": "user.message", "text": prompt},
        )
        for ev in stream:
            if ev.type == "agent.message.delta": chunks.append(ev.delta.text)
            elif ev.type == "agent.message.stop": break
    return "".join(chunks)

import anthropic

RESEARCH_SYSTEM = """You are the Research Agent for Mustafa Furniturewala.
You gather, verify, and synthesize information from the web.

RESEARCH DOMAINS (in priority order):
1. Generative AI — new models, papers, Anthropic news, Claude updates
2. EdTech — Coursera strategy, competitors (Udemy, LinkedIn Learning, Pluralsight, edX)
3. General tech — VP Engineering relevant: platform engineering, LLM infrastructure
4. Local — Sunnyvale 94087 events, safety, school news for Pinewood and Stratford
5. Personal brand — mentions of "Mustafa Furniturewala" across web, LinkedIn, GitHub

OUTPUT REQUIREMENTS:
- Bullet-point summaries, max 3 bullets per story
- Include source URL for every item
- Flag as [NEW] if not seen in last 7 days
- Flag as [URGENT] if time-sensitive (safety, breaking news)
- Rate importance: ⭐ nice-to-know, ⭐⭐ useful, ⭐⭐⭐ must-read

PRESS MENTION PROTOCOL:
If you find any mention of "Mustafa Furniturewala" or "mustafaf":
1. Record the URL, title, publication date, context
2. Return it in the mentions[] field of your JSON output
3. If it is positive, suggest a blog post title based on it

EDTECH COMPETITOR TRACKING:
For Coursera competitors, look for: product launches, pricing changes,
executive moves, partnership announcements, earnings reports."""

research_agent = client.beta.agents.create(
    model="claude-haiku-3-5",  # Cost-efficient for search-heavy tasks
    name="research-agent",
    system=RESEARCH_SYSTEM,
    tools=[
        {"type": "bash"},
        {"type": "text_editor"},
    ],
    mcp_servers=[
        {"name": "web-search", "url": "http://localhost:8001"},
        {"name": "web-fetch", "url": "http://localhost:8002"},
    ],
    max_tokens=8192,
)

def run_genai_digest(client, agent_id, env_id):
    """Daily GenAI digest — replaces daily-genai-digest Cowork task."""
    session = client.beta.sessions.create(agent_id=agent_id, environment_id=env_id)
    prompt = """Research today's top generative AI news:
1. Search for "generative AI news today", "Claude Anthropic", "GPT OpenAI", "Gemini Google"
2. Identify top 5-7 most significant developments
3. For each: brief summary, source URL, importance rating
4. Highlight anything directly relevant to my work at Coursera
5. Note any model releases, pricing changes, or API updates"""

    chunks = []
    with client.beta.sessions.events.stream(session.id) as stream:
        client.beta.sessions.events.send(
            session_id=session.id,
            event={"type": "user.message", "text": prompt},
        )
        for ev in stream:
            if ev.type == "agent.message.delta": chunks.append(ev.delta.text)
            elif ev.type == "agent.message.stop": break
    return "".join(chunks)

import anthropic

BROWSER_SYSTEM = """You are the Browser Agent for Mustafa Furniturewala.
You navigate real websites using Chrome automation.

KNOWN BOOKMARKED TASKS:

FAIRBRAE TENNIS BOOKING:
- Target: Fairbrae Recreation Center, Sunnyvale CA
- Preferred: Court 2, Sundays, early morning if available
- Fallback: Court 1, any Sunday slot
- If fully booked: report back, do NOT book an alternative day
- Booking portal URL: stored in your bash environment as $FAIRBRAE_URL

NIKE PEGASUS 42 DEAL TRACKER:
- Target price threshold: $100 or below (currently tracking)
- Check nike.com and runningwarehouse.com
- Size: 11.5 US men's
- If price drops below threshold: mark as DEAL_FOUND in output
- Check frequency: weekly (you'll be called automatically)

TRIP RESEARCH (Iceland, Turkey):
- When asked for restaurant/activity recs, use TripAdvisor and Google Maps
- Filter by: highly rated (4.0+), open during trip dates
- For Iceland: Reykjavik focus unless specific region requested
- For Turkey: Istanbul focus unless specific region requested

OPERATING RULES:
- Never enter payment information on any site
- If a site requires login, check bash environment for stored credentials
- Screenshot key pages for verification before submitting forms
- If booking fails 3 times, report failure, don't retry
- Return confirmation numbers when available"""

browser_agent = client.beta.agents.create(
    model="claude-sonnet-4-6",  # Sonnet for complex navigation decisions
    name="browser-agent",
    system=BROWSER_SYSTEM,
    tools=[{"type": "bash"}],
    mcp_servers=[{
        "name": "chrome",
        "url": "http://localhost:9222",  # Chrome DevTools Protocol MCP
    }],
    max_tokens=4096,
)

def book_tennis_court(client, agent_id, env_id, target_date: str):
    """Book Fairbrae Court 2 for the given Sunday date."""
    session = client.beta.sessions.create(agent_id=agent_id, environment_id=env_id)
    prompt = f"""Book Fairbrae Tennis Court 2 for {target_date} (Sunday).
Steps:
1. Navigate to the Fairbrae booking portal ($FAIRBRAE_URL)
2. Log in with stored credentials
3. Find Court 2 availability for {target_date}
4. Select the earliest available morning slot (before 10am if possible)
5. Complete the booking (no payment required for residents)
6. Screenshot the confirmation page
7. Return the confirmation number and time slot"""

    chunks = []
    with client.beta.sessions.events.stream(session.id) as stream:
        client.beta.sessions.events.send(
            session_id=session.id,
            event={"type": "user.message", "text": prompt},
        )
        for ev in stream:
            if ev.type == "agent.message.delta": chunks.append(ev.delta.text)
            elif ev.type == "agent.message.stop": break
    return "".join(chunks)

CODE_SYSTEM = """You are the Code and Deploy Agent for Mustafa Furniturewala.
You manage code, content, and deployments.

REPOSITORIES:
- Personal blog: github.com/mustafafurniturewala/blog (Vercel-deployed)
- Home automation scripts: ~/agent-network/ (local)
- Config files: ~/.config/agent-network/

BLOG WORKFLOW (for press mentions):
1. Receive press mention details from Research Agent
2. Draft a short blog post (300-500 words) in Mustafa's voice
3. Create a new MDX file in blog/content/press/
4. Include: mention source, date, context, Mustafa's brief comment
5. Run git add, commit, push
6. Vercel will auto-deploy on push to main
7. Return the deployed URL

DEPLOYMENT RULES:
- Always run tests before deploying: npm test or pytest
- Never deploy with failing tests
- Use descriptive commit messages
- Tag releases for major changes: git tag v{date}-{feature}
- If deploy fails, roll back immediately: git revert HEAD

FILE MANAGEMENT:
- Write clean, documented code
- Use Mustafa's preferred patterns (check existing files for style)
- Log all file changes to ~/.config/agent-network/file_changes.log"""

code_agent = client.beta.agents.create(
    model="claude-sonnet-4-6",
    name="code-agent",
    system=CODE_SYSTEM,
    tools=[
        {"type": "bash"},
        {"type": "text_editor"},
    ],
    mcp_servers=[
        {"name": "github", "url": "http://localhost:8010"},
        {"name": "vercel", "url": "http://localhost:8011"},
    ],
    max_tokens=8192,
)

DISPATCHER_SYSTEM = """You are the Master Dispatcher for Mustafa Furniturewala's agent network.
You classify incoming requests and route them to specialist agents.

AVAILABLE SPECIALISTS AND THEIR CAPABILITIES:
1. calendar-agent: Google Calendar CRUD, conflict detection, family scheduling
2. email-agent: Gmail search, read, draft (no send), subscription alerts
3. notification-agent: iMessage send (rate-limited), family alerts
4. research-agent: Web search, news digest, press mentions, competitor tracking
5. browser-agent: Chrome automation, Fairbrae booking, deal tracking, trip research
6. code-agent: File system, git, Vercel deploy, blog posts
7. kids-learning-agent: Content for Rehaan (Stratford) and Zara (Pinewood, 6th grade)

ROUTING RULES:
"schedule" / "calendar" / "event" / "conflict" → calendar-agent
"email" / "gmail" / "draft" / "reply" / "inbox" / "subscription" → email-agent
"send message" / "text" / "alert" / "urgent" / "notify family" → notification-agent
"news" / "digest" / "research" / "find" / "search" / "mentions" → research-agent
"book" / "tennis" / "Fairbrae" / "nike" / "price" / "website" / "browse" → browser-agent
"deploy" / "blog" / "code" / "github" / "vercel" / "file" → code-agent
"Rehaan" / "Zara" / "school" / "learning" / "math" / "kids" → kids-learning-agent

PARALLEL EXECUTION TRIGGERS:
When a request involves multiple domains, run agents concurrently:
- "Morning digest" → [research-agent, calendar-agent, email-agent] in parallel
- "Trip planning" → [research-agent, calendar-agent, browser-agent] in parallel
- "Weekly review" → [email-agent, research-agent, calendar-agent] in parallel

COMMAND DISPATCHER PROTOCOL (from iMessage //claude commands):
Parse the command after //claude and route accordingly.
Examples:
  "//claude book tennis Sunday" → browser-agent (Fairbrae booking)
  "//claude any emails from Coursera leadership?" → email-agent
  "//claude what's on the family calendar this week?" → calendar-agent
  "//claude news on Anthropic today" → research-agent

AGGREGATION RULES:
- Combine specialist outputs into a single coherent response
- Lead with the most actionable information
- Use clear section headers per specialist domain
- If a specialist failed, note it but don't let it block other results
- Maximum response length: 2000 characters for iMessage, unlimited for email

ERROR HANDLING:
If a specialist fails: log the error, continue with others, note the failure in output.
If all specialists fail: return a brief error with manual fallback instructions.
Never surface raw stack traces to the user.

OUTPUT FORMAT:
{
  "routing_decision": [{"agent": str, "reason": str, "priority": "parallel|sequential"}],
  "results": {: },
  "aggregated_summary": str,
  "errors": [{"agent": str, "error": str}],
  "actions_taken": [str]
}"""

import anthropic
import asyncio
import json
import logging
from typing import Optional
from dataclasses import dataclass
from concurrent.futures import ThreadPoolExecutor

logger = logging.getLogger("dispatcher")

@dataclass
class AgentConfig:
    agent_id: str
    environment_id: str
    name: str
    timeout_seconds: int = 120

class AgentDispatcher:
    """Central dispatcher for the Furniturewala agent network."""

    def __init__(self, config_path: str = "config/agents.json"):
        self.client = anthropic.Anthropic()
        self.agents: dict[str, AgentConfig] = {}
        self.dispatcher_agent_id: Optional[str] = None
        self.dispatcher_env_id: Optional[str] = None
        self._load_config(config_path)

    def _load_config(self, config_path: str):
        with open(config_path) as f:
            config = json.load(f)
        self.dispatcher_agent_id = config["dispatcher_agent_id"]
        self.dispatcher_env_id = config["dispatcher_env_id"]
        for name, cfg in config["agents"].items():
            self.agents[name] = AgentConfig(
                agent_id=cfg["agent_id"],
                environment_id=cfg["environment_id"],
                name=name,
                timeout_seconds=cfg.get("timeout_seconds", 120),
            )

    def _run_specialist(self, agent_name: str, prompt: str) -> dict:
        """Run a single specialist agent and return its result."""
        cfg = self.agents.get(agent_name)
        if not cfg:
            return {"error": f"Unknown agent: {agent_name}"}

        try:
            session = self.client.beta.sessions.create(
                agent_id=cfg.agent_id,
                environment_id=cfg.environment_id,
            )
            chunks = []
            with self.client.beta.sessions.events.stream(session.id) as stream:
                self.client.beta.sessions.events.send(
                    session_id=session.id,
                    event={"type": "user.message", "text": prompt},
                )
                for event in stream:
                    if event.type == "agent.message.delta":
                        chunks.append(event.delta.text)
                    elif event.type == "agent.message.stop":
                        break
            return {"agent": agent_name, "result": "".join(chunks), "status": "ok"}

        except Exception as e:
            logger.error(f"Specialist {agent_name} failed: {e}")
            return {"agent": agent_name, "error": str(e), "status": "failed"}

    def run_parallel(self, agent_tasks: list[tuple[str, str]]) -> list[dict]:
        """Run multiple agents concurrently. agent_tasks = [(agent_name, prompt), ...]"""
        with ThreadPoolExecutor(max_workers=min(len(agent_tasks), 6)) as executor:
            futures = [
                executor.submit(self._run_specialist, name, prompt)
                for name, prompt in agent_tasks
            ]
            return [f.result() for f in futures]

    def dispatch(self, user_request: str, source: str = "scheduler") -> dict:
        """Main dispatch entry point. Routes request, runs specialists, aggregates."""
        logger.info(f"Dispatching [{source}]: {user_request[:80]}...")

        # Step 1: Ask dispatcher agent to classify and plan
        dispatch_session = self.client.beta.sessions.create(
            agent_id=self.dispatcher_agent_id,
            environment_id=self.dispatcher_env_id,
        )
        plan_prompt = f"""Request from {source}: {user_request}

Classify this request and return a JSON routing plan:
{{
  "agents_needed": ["agent1", "agent2"],
  "execution": "parallel" or "sequential",
  "prompts": {{"agent1": "specific prompt for agent1", "agent2": "..."}}
}}"""

        plan_chunks = []
        with self.client.beta.sessions.events.stream(dispatch_session.id) as stream:
            self.client.beta.sessions.events.send(
                session_id=dispatch_session.id,
                event={"type": "user.message", "text": plan_prompt},
            )
            for ev in stream:
                if ev.type == "agent.message.delta": plan_chunks.append(ev.delta.text)
                elif ev.type == "agent.message.stop": break

        plan_text = "".join(plan_chunks)

        try:
            j_start = plan_text.find("{"); j_end = plan_text.rfind("}") + 1
            plan = json.loads(plan_text[j_start:j_end])
        except:
            return {"error": "Dispatcher failed to produce routing plan", "raw": plan_text}

        # Step 2: Execute specialist agents
        agent_tasks = [
            (agent, plan["prompts"][agent])
            for agent in plan["agents_needed"]
            if agent in self.agents
        ]

        if plan.get("execution") == "parallel":
            results = self.run_parallel(agent_tasks)
        else:
            results = [self._run_specialist(n, p) for n, p in agent_tasks]

        return {
            "source": source,
            "request": user_request,
            "plan": plan,
            "results": results,
            "errors": [r for r in results if r.get("status") == "failed"],
        }

import re
from dispatcher.main import AgentDispatcher

COMMAND_PREFIX = "//claude"
MUSTAFA_NUMBER = "+1XXXXXXXXXX"  # Mustafa's own number for self-commands

def check_imessage_commands(dispatcher: AgentDispatcher):
    """Called hourly by cron. Reads recent iMessages and processes //claude commands."""
    # In production, use iMessage MCP to get unread messages
    # Here we show the pattern
    recent_messages = get_recent_imessages()  # via iMessage MCP

    for msg in recent_messages:
        if re.search(rf"^{re.escape(COMMAND_PREFIX)}\s+", msg["text"], re.IGNORECASE):
            command = msg["text"][len(COMMAND_PREFIX):].strip()
            sender = msg["sender"]

            # Only process commands from Mustafa himself or immediate family
            if sender not in [MUSTAFA_NUMBER]:
                continue

            result = dispatcher.dispatch(command, source="imessage")
            send_imessage_reply(sender, format_result(result))

def format_result(result: dict) -> str:
    """Format dispatcher result for iMessage (160-char friendly)."""
    ok_results = [r for r in result["results"] if r.get("status") == "ok"]
    if not ok_results:
        return "⚠️ All agents failed. Check logs."
    # Take the first successful result's first 300 chars
    return ok_results[0]["result"][:300]

from dispatcher.main import AgentDispatcher
from datetime import datetime

def run_morning_digest(dispatcher: AgentDispatcher):
    """Full morning digest: calendar + email + research in parallel."""
    today = datetime.now().strftime("%A, %B %d, %Y")

    agent_tasks = [
        ("calendar-agent",
         f"Generate the family calendar digest for today ({today}) and the next 3 days. "
         "Highlight any conflicts, school events for Rehaan and Zara, and new events."),

        ("email-agent",
         "Scan inbox for unread emails since yesterday 6pm. "
         "Summarize high-priority items only. Flag anything requiring action today."),

        ("research-agent",
         "Run the morning news digest: top 5 GenAI news, top 3 EdTech/Coursera news, "
         "any Sunnyvale local alerts, any new press mentions of Mustafa Furniturewala."),
    ]

    # Run all three in parallel — typically completes in ~30 seconds
    results = dispatcher.run_parallel(agent_tasks)

    # Assemble the morning email
    sections = [f"# Good morning, Mustafa! — {today}\n"]

    for r in results:
        if r["status"] == "ok":
            sections.append(f"\n## {r['agent'].replace('-', ' ').title()}\n")
            sections.append(r["result"])
        else:
            sections.append(f"\n## {r['agent']} — ⚠️ Failed: {r.get('error','unknown')}\n")

    return "\n".join(sections)

# Furniturewala Agent Network — Scheduled Task Configuration
# Timezone: America/Los_Angeles (Sunnyvale, CA)
# All times in 24h local time unless noted

defaults:
  timezone: America/Los_Angeles
  max_retries: 3
  retry_delay_seconds: 60
  timeout_seconds: 180

tasks:

  # ─── COMMAND PROCESSING ───────────────────────────────────────────────
  - id: command-dispatcher
    agent: dispatcher
    schedule: {type: interval, minutes: 60}
    prompt: "Check for new iMessage //claude commands and execute them."
    output: log

  - id: personal-assistant-check-in
    agent: dispatcher
    schedule: {type: interval, minutes: 120}
    prompt: "Run consolidated sweep: check calendar for new events, check email for urgent items, check any pending notifications."
    output: log

  # ─── EMAIL ────────────────────────────────────────────────────────────
  - id: hourly-email-drafter
    agent: email-agent
    schedule: {type: interval, hours: 4}
    prompt: "Run a 4-hour email sweep. Classify unread threads, draft Priority 1 and 2 responses."
    output: log

  - id: subscription-charge-alert
    agent: email-agent
    schedule: {type: cron, day_of_week: mon, hour: 8, minute: 0}
    prompt: "Scan Gmail for subscription renewal and billing emails from the past 7 days. List all services, amounts, and upcoming renewal dates."
    output: email
    email_to: mustafaf@gmail.com

  # ─── NEWS RESEARCH ────────────────────────────────────────────────────
  - id: daily-news-digest
    agent: research-agent
    schedule: {type: cron, hour: 6, minute: 30}
    prompt: "Gather today's top 10 news stories across tech, business, and world news. Summarize each in 2-3 bullets with source URLs."
    output: email
    email_to: [mustafaf@gmail.com, zeesha@gmail.com]

  - id: daily-genai-digest
    agent: research-agent
    schedule: {type: cron, hour: 7, minute: 0}
    prompt: "Research today's top 7 generative AI developments. Focus on model releases, API changes, research papers, and business news. Highlight Coursera-relevant items."
    output: email
    email_to: mustafaf@gmail.com

  - id: daily-edtech-news-digest
    agent: research-agent
    schedule: {type: cron, hour: 7, minute: 30}
    prompt: "Search for today's edtech news: Coursera announcements, competitor activity (Udemy, LinkedIn Learning, edX, Pluralsight). Include any executive moves, product launches, or pricing changes."
    output: email
    email_to: mustafaf@gmail.com

  - id: weekly-claude-updates
    agent: research-agent
    schedule: {type: cron, day_of_week: fri, hour: 17, minute: 0}
    prompt: "Summarize this week's Claude and Anthropic updates: new model capabilities, API changes, pricing, policy updates, and notable community use cases."
    output: email
    email_to: mustafaf@gmail.com

  # ─── FAMILY & KIDS ────────────────────────────────────────────────────
  - id: school-events-digest
    agent: research-agent
    schedule: {type: cron, hour: 6, minute: 45}
    prompt: "Check for school events and news: Pinewood School (Zara, 6th grade) and Stratford School (Rehaan). Look for upcoming events, school news, sports schedules, and important dates."
    output: email
    email_to: mustafaf@gmail.com

  - id: kids-daily-news-digest
    agent: kids-learning-agent
    schedule: {type: cron, hour: 7, minute: 15}
    prompt: "Create a fun, age-appropriate daily news summary for Rehaan and Zara. Focus on science discoveries, sports, interesting world events. Keep it engaging and educational."
    output: email
    email_to: mustafaf@gmail.com

  - id: zaras-daily-learning-email
    agent: kids-learning-agent
    schedule: {type: cron, hour: 15, minute: 30}
    prompt: "Generate 3 personalized math problems and 2 science questions for Zara (6th grade, Pinewood). Match California 6th grade Common Core standards. Include hints. Make it engaging."
    output: email
    email_to: mustafaf@gmail.com

  - id: family-dinner-conversation
    agent: research-agent
    schedule: {type: cron, hour: 16, minute: 0}
    prompt: "Generate 3 interesting family dinner conversation topics for tonight. Mix: one topic for kids (fun science or history), one for Mustafa and Zeesha (current events), one that involves everyone (thought experiment or hypothetical). Keep it light and engaging."
    output: imessage
    imessage_to: self

  # ─── SAFETY & LOCAL ───────────────────────────────────────────────────
  - id: urgent-news-midday
    agent: research-agent
    schedule: {type: cron, hour: 12, minute: 0}
    prompt: "Check for urgent local news affecting Sunnyvale 94087: safety alerts, traffic incidents, weather warnings, school closures. Only report if there is something genuinely urgent — skip if nothing notable."
    output: conditional_imessage
    condition: result_contains_urgent
    imessage_to: self

  - id: local-events-digest
    agent: research-agent
    schedule: {type: cron, day_of_week: fri, hour: 18, minute: 0}
    prompt: "Find interesting local events near Sunnyvale 94087 for the upcoming weekend: family activities, food events, outdoor activities. Include dates, times, locations."
    output: email
    email_to: mustafaf@gmail.com

  # ─── CALENDAR ─────────────────────────────────────────────────────────
  - id: family-calendar-event-notifier
    agent: calendar-agent
    schedule: {type: cron, hour: 8, minute: 0}
    prompt: "Check family calendar for newly added events since yesterday. Summarize new events, flag conflicts, and notify relevant family members."
    output: email
    email_to: mustafaf@gmail.com

  # ─── BROWSER AUTOMATION ───────────────────────────────────────────────
  - id: weekly-fairbrae-tennis
    agent: browser-agent
    schedule: {type: cron, day_of_week: sun, hour: 6, minute: 0}
    prompt: "Book Fairbrae Tennis Court 2 for next Sunday. If unavailable, try Court 1. Report booking confirmation."
    output: imessage
    imessage_to: self

  - id: nike-pegasus-42-deal-tracker
    agent: browser-agent
    schedule: {type: cron, day_of_week: wed, hour: 10, minute: 0}
    prompt: "Check Nike Pegasus 42 price (size 11.5 US men's) on nike.com and runningwarehouse.com. Alert if price is $100 or below."
    output: conditional_imessage
    condition: price_below_threshold
    imessage_to: self

  # ─── ONLINE PRESENCE ──────────────────────────────────────────────────
  - id: online-presence-monitor
    agent: research-agent
    schedule: {type: cron, hour: 9, minute: 0}
    prompt: "Search for online mentions of 'Mustafa Furniturewala' across web, LinkedIn, GitHub, Twitter/X. Summarize any new mentions found. Also suggest one blog post topic based on current GenAI trends."
    output: log

  - id: check-press-mentions
    agent: research-agent
    schedule: {type: cron, day_of_week: mon, hour: 9, minute: 30}
    prompt: "Find any new press mentions, podcast appearances, or article quotes featuring Mustafa Furniturewala from the past week. If found, prepare a blog post draft."
    output: log_and_draft

import yaml
import logging
import time
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.interval import IntervalTrigger
from dispatcher.main import AgentDispatcher

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    handlers=[
        logging.StreamHandler(),
        logging.FileHandler("logs/scheduler.log"),
    ]
)
logger = logging.getLogger("scheduler")

class AgentTaskRunner:
    def __init__(self, config_path: str = "config/tasks.yaml"):
        self.dispatcher = AgentDispatcher()
        self.scheduler = BlockingScheduler(timezone="America/Los_Angeles")
        self.config = self._load_config(config_path)

    def _load_config(self, path: str) -> dict:
        with open(path) as f:
            return yaml.safe_load(f)

    def _build_trigger(self, schedule_cfg: dict):
        stype = schedule_cfg["type"]
        if stype == "cron":
            kwargs = {k: v for k, v in schedule_cfg.items() if k != "type"}
            return CronTrigger(timezone="America/Los_Angeles", **kwargs)
        elif stype == "interval":
            kwargs = {k: v for k, v in schedule_cfg.items() if k != "type"}
            return IntervalTrigger(**kwargs)
        raise ValueError(f"Unknown schedule type: {stype}")

    def _make_job(self, task: dict):
        """Create a job function for a task config."""
        task_id = task["id"]
        prompt = task["prompt"]
        max_retries = task.get("max_retries", self.config["defaults"]["max_retries"])
        retry_delay = task.get("retry_delay_seconds", self.config["defaults"]["retry_delay_seconds"])

        def job():
            for attempt in range(1, max_retries + 1):
                try:
                    logger.info(f"Running task: {task_id} (attempt {attempt})")
                    result = self.dispatcher.dispatch(prompt, source=task_id)
                    logger.info(f"Task {task_id} completed: {str(result)[:200]}")
                    return
                except Exception as e:
                    logger.error(f"Task {task_id} attempt {attempt} failed: {e}")
                    if attempt < max_retries:
                        wait = retry_delay * (2 ** (attempt - 1))  # Exponential backoff
                        logger.info(f"Retrying in {wait}s...")
                        time.sleep(wait)
            logger.error(f"Task {task_id} FAILED after {max_retries} attempts")
        return job

    def start(self):
        """Register all tasks and start the scheduler."""
        for task in self.config["tasks"]:
            trigger = self._build_trigger(task["schedule"])
            job_fn = self._make_job(task)
            self.scheduler.add_job(job_fn, trigger, id=task["id"], name=task["id"])
            logger.info(f"Registered task: {task['id']}")

        logger.info(f"Starting scheduler with {len(self.config['tasks'])} tasks")
        self.scheduler.start()

if __name__ == "__main__":
    runner = AgentTaskRunner()
    runner.start()

import anthropic
import json

client = anthropic.Anthropic()

MEMORY_STORES = [
    {"name": "user-profile",       "description": "Mustafa's preferences, habits, and identity"},
    {"name": "research-seen",      "description": "News items already sent to avoid duplicates"},
    {"name": "calendar-state",     "description": "Last known calendar state for diff detection"},
    {"name": "email-context",      "description": "Email threads in progress, draft states"},
    {"name": "notification-log",   "description": "iMessage send history for rate limiting"},
    {"name": "press-mentions",     "description": "Press mentions found, blog post status"},
    {"name": "trip-context",       "description": "Iceland and Turkey trip details, preferences"},
    {"name": "kids-learning",      "description": "Zara progress, topics covered, Rehaan interests"},
]

store_ids = {}
for store_cfg in MEMORY_STORES:
    store = client.beta.memory_stores.create(
        name=store_cfg["name"],
        description=store_cfg["description"],
    )
    store_ids[store_cfg["name"]] = store.id
    print(f"Created store '{store_cfg['name']}': {store.id}")

# Save store IDs to config
with open("config/memory_stores.json", "w") as f:
    json.dump(store_ids, f, indent=2)
print("Memory stores configured.")

import anthropic
import json
from datetime import datetime

client = anthropic.Anthropic()

def initialize_user_profile(store_id: str):
    """Populate the user profile memory store with Mustafa's details."""

    profile_doc = {
        "identity": {
            "name": "Mustafa Furniturewala",
            "email": "mustafaf@gmail.com",
            "role": "VP Engineering at Coursera",
            "location": "Sunnyvale, CA 94087",
        },
        "family": {
            "spouse": "Zeesha",
            "children": [
                {"name": "Rehaan", "school": "Stratford"},
                {"name": "Zara", "school": "Pinewood", "grade": 6},
            ],
        },
        "communication_preferences": {
            "email_style": "Professional but direct. Short paragraphs. Sign as Mustafa.",
            "imessage_threshold": "Only genuinely urgent or actionable items",
            "news_preference": "GenAI and edtech primary. Bullet summaries, max 3 per story.",
        },
        "interests": [
            "Generative AI / LLMs",
            "Education technology",
            "Tennis (plays at Fairbrae Recreation Center, Court 2 preferred)",
            "Running (Nike Pegasus 42, size 11.5 US mens, target price $100)",
            "Family travel (upcoming: Iceland, Turkey)",
        ],
        "active_trips": [
            {"destination": "Iceland", "status": "planning"},
            {"destination": "Turkey", "status": "planning"},
        ],
        "last_updated": datetime.now().isoformat(),
    }

    client.beta.memory_stores.documents.create(
        memory_store_id=store_id,
        content=json.dumps(profile_doc, indent=2),
        title="Mustafa Furniturewala — User Profile",
        metadata={"type": "user_profile", "version": "1.0"},
    )
    print("User profile initialized in memory store")

import anthropic
import hashlib
import json
from datetime import datetime, timedelta

class ResearchDeduplicator:
    """Track seen news items to prevent duplicate delivery."""

    def __init__(self, client: anthropic.Anthropic, store_id: str):
        self.client = client
        self.store_id = store_id

    def _make_key(self, url: str) -> str:
        return hashlib.md5(url.encode()).hexdigest()[:12]

    def is_seen(self, url: str) -> bool:
        """Check if a URL has been sent in the last 7 days."""
        key = self._make_key(url)
        try:
            results = self.client.beta.memory_stores.documents.list(
                memory_store_id=self.store_id,
                query=key,
            )
            for doc in results.data:
                meta = json.loads(doc.metadata or "{}")
                if meta.get("url_hash") == key:
                    seen_date = datetime.fromisoformat(meta["seen_at"])
                    if (datetime.now() - seen_date).days < 7:
                        return True
        except Exception:
            pass  # On error, don't block delivery
        return False

    def mark_seen(self, url: str, title: str):
        """Record that a URL was delivered."""
        key = self._make_key(url)
        self.client.beta.memory_stores.documents.create(
            memory_store_id=self.store_id,
            content=f"Seen: {title} ({url})",
            title=f"seen-{key}",
            metadata=json.dumps({
                "url_hash": key,
                "url": url,
                "title": title,
                "seen_at": datetime.now().isoformat(),
            }),
        )

    def filter_new(self, items: list[dict]) -> list[dict]:
        """Filter a list of {url, title, ...} items to only new ones."""
        new_items = []
        for item in items:
            if not self.is_seen(item["url"]):
                new_items.append(item)
                self.mark_seen(item["url"], item.get("title", ""))
        return new_items

import anthropic

client = anthropic.Anthropic()

# Agent with multiple MCP servers for full comms access
comms_agent = client.beta.agents.create(
    model="claude-sonnet-4-6",
    name="comms-agent",
    system="You manage communications for Mustafa Furniturewala.",
    tools=[{"type": "bash"}],
    mcp_servers=[
        # Gmail via Google's official MCP endpoint
        {
            "name": "gmail",
            "url": "https://mcp.googleapis.com/gmail/v1",
            "auth": {
                "type": "oauth2",
                "credentials_env": "GOOGLE_GMAIL_CREDENTIALS",
            },
        },
        # Google Calendar
        {
            "name": "google-calendar",
            "url": "https://mcp.googleapis.com/calendar/v1",
            "auth": {
                "type": "oauth2",
                "credentials_env": "GOOGLE_CALENDAR_CREDENTIALS",
            },
        },
        # Local iMessage server (runs on Mustafa's Mac)
        {
            "name": "imessage",
            "url": "http://localhost:7777",
            "auth": {
                "type": "bearer",
                "token_env": "IMESSAGE_MCP_TOKEN",
            },
        },
    ],
    max_tokens=8192,
)

from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import sqlite3
import subprocess
import json
import os
from pathlib import Path

app = Server("imessage-mcp")

MESSAGES_DB = Path(os.path.expanduser("~/Library/Messages/chat.db"))

# ─── TOOL DEFINITIONS ─────────────────────────────────────────────────

@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="read_messages",
            description="Read recent iMessages from a contact or all unread",
            inputSchema={
                "type": "object",
                "properties": {
                    "contact": {"type": "string", "description": "Phone number or empty for all"},
                    "limit": {"type": "integer", "description": "Max messages to return", "default": 20},
                    "unread_only": {"type": "boolean", "default": True},
                },
            },
        ),
        Tool(
            name="send_message",
            description="Send an iMessage to a phone number",
            inputSchema={
                "type": "object",
                "required": ["recipient", "message"],
                "properties": {
                    "recipient": {"type": "string", "description": "Phone number in E.164 format"},
                    "message": {"type": "string", "description": "Message text to send"},
                },
            },
        ),
        Tool(
            name="search_messages",
            description="Search iMessage history for a keyword or command prefix",
            inputSchema={
                "type": "object",
                "required": ["query"],
                "properties": {
                    "query": {"type": "string"},
                    "hours_back": {"type": "integer", "default": 2},
                },
            },
        ),
    ]

# ─── TOOL IMPLEMENTATIONS ──────────────────────────────────────────────

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:

    if name == "read_messages":
        messages = _read_messages(
            contact=arguments.get("contact"),
            limit=arguments.get("limit", 20),
            unread_only=arguments.get("unread_only", True),
        )
        return [TextContent(type="text", text=json.dumps(messages, indent=2))]

    elif name == "send_message":
        result = _send_imessage(arguments["recipient"], arguments["message"])
        return [TextContent(type="text", text=json.dumps(result))]

    elif name == "search_messages":
        results = _search_messages(arguments["query"], arguments.get("hours_back", 2))
        return [TextContent(type="text", text=json.dumps(results, indent=2))]

    raise ValueError(f"Unknown tool: {name}")

def _read_messages(contact=None, limit=20, unread_only=True) -> list[dict]:
    """Query the macOS Messages SQLite database."""
    conn = sqlite3.connect(str(MESSAGES_DB))
    conn.row_factory = sqlite3.Row
    cursor = conn.cursor()

    query = """
        SELECT m.text, m.date, m.is_from_me,
               h.id as sender, m.is_read
        FROM message m
        JOIN handle h ON m.handle_id = h.ROWID
        WHERE m.text IS NOT NULL
    """
    params = []
    if contact:
        query += " AND h.id LIKE ?"; params.append(f"%{contact}%")
    if unread_only:
        query += " AND m.is_read = 0"
    query += f" ORDER BY m.date DESC LIMIT {limit}"

    cursor.execute(query, params)
    return [dict(row) for row in cursor.fetchall()]

def _send_imessage(recipient: str, message: str) -> dict:
    """Send via AppleScript — requires macOS Messages app access."""
    script = f'''
    tell application "Messages"
        set targetService to 1st service whose service type = iMessage
        set targetBuddy to buddy "{recipient}" of targetService
        send "{message}" to targetBuddy
    end tell
    '''
    try:
        subprocess.run(["osascript", "-e", script], check=True, capture_output=True)
        return {"status": "sent", "recipient": recipient}
    except subprocess.CalledProcessError as e:
        return {"status": "failed", "error": e.stderr.decode()}

def _search_messages(query: str, hours_back: int = 2) -> list[dict]:
    """Search messages containing query string in recent hours."""
    conn = sqlite3.connect(str(MESSAGES_DB))
    conn.row_factory = sqlite3.Row
    cursor = conn.cursor()
    # macOS stores dates as seconds since 2001-01-01
    cutoff = (datetime.now() - timedelta(hours=hours_back)).timestamp() - 978307200
    cursor.execute(
        "SELECT text, date FROM message WHERE text LIKE ? AND date > ? ORDER BY date DESC",
        (f"%{query}%", cutoff)
    )
    return [dict(row) for row in cursor.fetchall()]

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream, app.create_initialization_options())

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

# Load secrets from macOS Keychain (or 1Password CLI) into env vars
# 1Password CLI example:
export GOOGLE_GMAIL_CREDENTIALS=$(op read "op://Personal/gmail-mcp/credentials")
export GOOGLE_CALENDAR_CREDENTIALS=$(op read "op://Personal/calendar-mcp/credentials")
export IMESSAGE_MCP_TOKEN=$(op read "op://Personal/imessage-mcp/token")
export ANTHROPIC_API_KEY=$(op read "op://Personal/anthropic/api-key")

# Then start the MCP servers and agent scheduler
python mcp_servers/imessage_server.py &
python mcp_servers/web_search_server.py &
python scheduler/runner.py

from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import httpx
import json
import os

app = Server("web-search-mcp")
SERPER_KEY = os.environ["SERPER_API_KEY"]

@app.list_tools()
async def list_tools():
    return [
        Tool(
            name="web_search",
            description="Search the web and return top results with titles, snippets, and URLs",
            inputSchema={
                "type": "object",
                "required": ["query"],
                "properties": {
                    "query": {"type": "string"},
                    "num_results": {"type": "integer", "default": 10},
                    "time_range": {
                        "type": "string",
                        "enum": ["any", "d", "w", "m"],
                        "description": "Time range: any, d=day, w=week, m=month",
                    },
                },
            },
        ),
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "web_search":
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                "https://google.serper.dev/search",
                headers={"X-API-KEY": SERPER_KEY},
                json={
                    "q": arguments["query"],
                    "num": arguments.get("num_results", 10),
                    "tbs": arguments.get("time_range", "any"),
                },
            )
            data = resp.json()
            results = [
                {"title": r["title"], "url": r["link"], "snippet": r.get("snippet", "")}
                for r in data.get("organic", [])
            ]
        return [TextContent(type="text", text=json.dumps(results, indent=2))]

version: "3.9"

networks:
  agent-net:
    driver: bridge

services:

  scheduler:
    build:
      context: .
      dockerfile: docker/Dockerfile.scheduler
    container_name: agent-scheduler
    networks: [agent-net]
    environment:
      - ANTHROPIC_API_KEY
      - GOOGLE_GMAIL_CREDENTIALS
      - GOOGLE_CALENDAR_CREDENTIALS
      - SERPER_API_KEY
      - IMESSAGE_MCP_URL=http://host.docker.internal:7777
      - CHROME_MCP_URL=http://host.docker.internal:9222
      - TZ=America/Los_Angeles
    volumes:
      - ./config:/app/config:ro
      - ./logs:/app/logs
      - agent-memory:/app/memory
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "python", "-c", "import scheduler.runner"]
      interval: 60s
      timeout: 10s
      retries: 3

  web-search-mcp:
    build:
      context: .
      dockerfile: docker/Dockerfile.mcp
    container_name: web-search-mcp
    networks: [agent-net]
    environment:
      - SERPER_API_KEY
    ports:
      - "8001:8001"
    command: python mcp_servers/web_search_server.py
    restart: unless-stopped

volumes:
  agent-memory:

FROM python:3.12-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application code
COPY . .

# Create log directory
RUN mkdir -p logs

# Run as non-root for security
RUN useradd -r -u 1000 agent && chown -R agent:agent /app
USER agent

CMD ["python", "-m", "scheduler.runner"]

anthropic>=0.50.0
apscheduler>=3.10.0
pyyaml>=6.0
httpx>=0.26.0
mcp>=1.0.0
python-dotenv>=1.0.0
structlog>=24.0.0

"""
Furniturewala Agent Network — Main Entry Point
Bootstraps all agents, memory stores, and the scheduler.
Run once to provision, then scheduler takes over.
"""
import argparse
import json
import logging
import os
from pathlib import Path
import anthropic
from agents.calendar_agent import calendar_agent
from agents.email_agent import email_agent
from agents.notification_agent import notification_agent
from agents.research_agent import research_agent
from agents.browser_agent import browser_agent
from agents.code_agent import code_agent
from memory.setup import MEMORY_STORES
from scheduler.runner import AgentTaskRunner

logger = logging.getLogger("main")

def provision():
    """Create all agents, environments, and memory stores. Run once."""
    client = anthropic.Anthropic()
    config = {}

    print("Provisioning Furniturewala Agent Network...")

    # Create environments
    envs = {}
    for name in ["comms-env", "research-env", "browser-env", "code-env", "dispatcher-env"]:
        env = client.beta.environments.create(name=name)
        envs[name] = env.id
        print(f"  ✓ Environment: {name}")
    config["environments"] = envs

    # Register agents and their environments
    agent_configs = {
        "calendar-agent": {"agent_id": calendar_agent.id, "environment_id": envs["comms-env"]},
        "email-agent":    {"agent_id": email_agent.id,    "environment_id": envs["comms-env"]},
        "notification-agent": {"agent_id": notification_agent.id, "environment_id": envs["comms-env"]},
        "research-agent": {"agent_id": research_agent.id, "environment_id": envs["research-env"]},
        "browser-agent":  {"agent_id": browser_agent.id,  "environment_id": envs["browser-env"]},
        "code-agent":     {"agent_id": code_agent.id,     "environment_id": envs["code-env"]},
    }
    config["agents"] = agent_configs

    # Create memory stores
    store_ids = {}
    for store_cfg in MEMORY_STORES:
        store = client.beta.memory_stores.create(name=store_cfg["name"])
        store_ids[store_cfg["name"]] = store.id
        print(f"  ✓ Memory store: {store_cfg['name']}")
    config["memory_stores"] = store_ids

    # Save config for runtime use
    Path("config").mkdir(exist_ok=True)
    with open("config/agents.json", "w") as f:
        json.dump(config, f, indent=2)
    print("\n✓ Provisioning complete. Config saved to config/agents.json")

def run():
    """Start the scheduler daemon."""
    runner = AgentTaskRunner()
    print("Starting agent scheduler...")
    runner.start()

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("command", choices=["provision", "run"])
    args = parser.parse_args()

    if args.command == "provision":
        provision()
    elif args.command == "run":
        run()

# Model selection logic based on task complexity

MODEL_TIERS = {
    # Haiku: fast, cheap ($0.25/M input, $1.25/M output)
    # Use for: simple search, notification evaluation, classification
    "haiku": "claude-haiku-3-5",

    # Sonnet: balanced ($3/M input, $15/M output)
    # Use for: complex reasoning, drafting, multi-step tasks
    "sonnet": "claude-sonnet-4-6",
}

AGENT_MODELS = {
    "notification-agent": "haiku",   # Just evaluates and sends — simple
    "research-agent":     "haiku",   # Search + fetch + summarize — repetitive
    "kids-learning-agent":"sonnet",  # Pedagogical quality matters
    "calendar-agent":     "sonnet",  # Conflict detection needs reasoning
    "email-agent":        "sonnet",  # Draft quality must be high
    "browser-agent":      "sonnet",  # Complex navigation decisions
    "code-agent":         "sonnet",  # Code quality matters
    "dispatcher":         "haiku",   # Just routing — simple classification
}

# Estimated monthly cost at current usage patterns:
# Haiku tasks (research x365, notification x180, dispatcher x720): ~$8/month
# Sonnet tasks (email x90, calendar x30, browser x52, code x20): ~$18/month
# Total estimate: ~$26/month for full network automation

import structlog
import json
from datetime import datetime
from pathlib import Path

log = structlog.get_logger()

class AgentRunLogger:
    """Structured logging for agent task runs."""

    def __init__(self, log_dir: str = "logs"):
        Path(log_dir).mkdir(exist_ok=True)
        self.log_dir = log_dir
        self.runs = []

    def record_run(self, task_id: str, status: str,
                   duration_ms: int, tokens_used: int = 0, error: str = None):
        entry = {
            "timestamp": datetime.now().isoformat(),
            "task_id": task_id,
            "status": status,
            "duration_ms": duration_ms,
            "tokens_used": tokens_used,
            "error": error,
        }
        self.runs.append(entry)
        log.info("agent_run", **entry)

        # Write to daily log file
        log_file = Path(self.log_dir) / f"{datetime.now().strftime('%Y-%m-%d')}.jsonl"
        with open(log_file, "a") as f:
            f.write(json.dumps(entry) + "\n")

    def daily_summary(self) -> dict:
        """Summarize today's runs for the morning digest."""
        total = len(self.runs)
        failed = [r for r in self.runs if r["status"] == "failed"]
        total_tokens = sum(r["tokens_used"] for r in self.runs)
        return {
            "total_runs": total,
            "successful": total - len(failed),
            "failed": len(failed),
            "failed_tasks": [r["task_id"] for r in failed],
            "total_tokens": total_tokens,
            "success_rate": f"{((total - len(failed)) / total * 100):.1f}%" if total else "N/A",
        }

furniturewala-agent-network/
├── main.py                        # Entry point: provision | run
├── requirements.txt
├── docker-compose.yml
├── docker/
│   ├── Dockerfile.scheduler
│   └── Dockerfile.mcp
├── config/
│   ├── tasks.yaml                 # All scheduled task definitions
│   ├── agents.json                # Generated after provision
│   └── memory_stores.json         # Generated after provision
├── agents/
│   ├── __init__.py
│   ├── calendar_agent.py
│   ├── email_agent.py
│   ├── notification_agent.py
│   ├── research_agent.py
│   ├── browser_agent.py
│   ├── code_agent.py
│   └── kids_learning_agent.py
├── dispatcher/
│   ├── __init__.py
│   ├── main.py                    # AgentDispatcher class
│   └── imessage_listener.py
├── scheduler/
│   ├── __init__.py
│   └── runner.py                  # AgentTaskRunner + APScheduler
├── mcp_servers/
│   ├── imessage_server.py         # Custom iMessage MCP
│   └── web_search_server.py       # Custom web search MCP
├── memory/
│   ├── setup.py                   # Create memory stores
│   ├── profile.py                 # User profile initialization
│   └── deduplication.py           # Research dedup logic
├── monitoring/
│   └── logger.py                  # Structured logging
└── logs/                          # Daily JSONL log files

# 1. Clone and install
git clone https://github.com/mustafafurniturewala/agent-network
cd agent-network
python -m venv venv && source venv/bin/activate
pip install -r requirements.txt

# 2. Load secrets (1Password CLI example)
source scripts/load_secrets.sh

# 3. Provision agents, environments, and memory stores (run once)
python main.py provision

# 4. Start MCP servers as background processes
python mcp_servers/imessage_server.py &
python mcp_servers/web_search_server.py &

# 5. Test with a single task run
python -c "
from dispatcher.main import AgentDispatcher
d = AgentDispatcher()
result = d.dispatch('What is on the family calendar this week?', source='test')
print(result)
"

# 6. Start the full scheduler daemon
python main.py run

# OR with Docker
docker-compose up -d