Agent Lifecycle

Agent Lifecycle

Overview

This document describes the full lifecycle of an agent: how the webhook handler triggers automatic creation via Rivet, how agents are reused, how they sleep/wake, and how they terminate.

Canonical reference: Rivet Actor Model contains the full implementation with code. This document focuses on the flow and state transitions.

What Manages Agents?

Rivet Runtime manages the agents. The FastAPI gateway is a thin router that does nothing but:

1. Receive webhook
2. Call rivet.get_actor(ShoppingAgentActor, user_id)
3. Send message to actor
4. Return 200 OK

Rivet handles everything else: creation, routing, concurrency, sleep/wake, termination, state persistence.

Flow: New User Texts

sequenceDiagram
    actor User
    participant WA as WhatsApp
    participant API as FastAPI Gateway
    participant RV as Rivet Runtime
    participant ACTOR as ShoppingAgentActor
    participant PI as Pi Agent
    participant MCP as @shopify/dev-mcp

    User->>WA: "Hi, show me red shoes"
    WA->>API: POST /webhook
    API-->>WA: 200 OK (instant)

    API->>RV: get_actor(ShoppingAgentActor, "+94771234567")

    Note over RV: user_id NOT in cluster<br/>→ CREATE new actor

    RV->>ACTOR: Instantiate actor
    ACTOR->>ACTOR: on_init()
    ACTOR->>MCP: Initialize Shopify MCP client
    ACTOR->>PI: Create Pi Agent + register tools
    ACTOR-->>RV: Ready

    RV-->>API: actor reference
    API->>ACTOR: on_message("Hi, show me red shoes")

    ACTOR->>ACTOR: Rate limit check
    ACTOR->>ACTOR: Injection check
    ACTOR->>PI: agent.prompt(text)
    PI->>MCP: search_shop_catalog("red shoes")
    MCP-->>PI: Products
    PI-->>ACTOR: Response
    ACTOR->>ACTOR: Save state (durable)
    ACTOR-->>API: Response
    API->>WA: Send WhatsApp reply
    WA-->>User: "Found 5 products..."

    Note over ACTOR: Sleeps after 5min idle

Flow: Returning User Texts

sequenceDiagram
    actor User
    participant WA as WhatsApp
    participant API as FastAPI Gateway
    participant RV as Rivet Runtime
    participant ACTOR as ShoppingAgentActor

    User->>WA: "Add the first one to cart"
    WA->>API: POST /webhook
    API-->>WA: 200 OK

    API->>RV: get_actor(ShoppingAgentActor, user_id)

    alt Actor is sleeping
        RV->>ACTOR: Wake (on_wake)
        ACTOR->>ACTOR: Reconnect MCP client
    else Actor is running
        Note over RV: Queue message<br/>(serialized execution)
    end

    RV-->>API: actor reference
    API->>ACTOR: on_message(text)
    Note over ACTOR: Has full conversation history<br/>(from durable state)
    ACTOR-->>API: Response
    API->>WA: Send reply
    WA-->>User: "Added to cart!"

State Lifecycle

stateDiagram-v2
    [*] --> NotExists: User has never texted

    NotExists --> Creating: First webhook
    Creating --> Running: on_init() complete
    Running --> Running: on_message() serialized
    Running --> Sleeping: 5 min idle
    Sleeping --> Running: Wake on message (on_wake)
    Sleeping --> Terminated: 1 hr idle
    Terminated --> NotExists: Destroyed
    Terminated --> Creating: Next message recreates

What Rivet Gives Us

Actor Implementation Sketch

@rivet.actor
class ShoppingAgentActor(Actor):
    class Config:
        idle_timeout_seconds = 300         # Sleep after 5 min
        termination_timeout_seconds = 3600 # Terminate after 1 hr
        durable = True

    async def on_init(self):
        """Create Pi Agent + register tools."""
        self.mcp_client = ShopifyMCPClient(...)
        await self.mcp_client.initialize()
        self.agent = Agent(AgentOptions(...))
        self.agent.set_tools(self.mcp_client.get_tools())

    async def on_message(self, text: str) -> str:
        """Called for each incoming message. Serialized by Rivet."""
        if not self._check_rate_limit():
            return "⏳ Slow down please."
        if not self._check_injection(text):
            return "I only help with shopping."

        response_parts = []
        def handler(event):
            if event.type == "message_update":
                for c in event.message.content:
                    if c.type == "text":
                        response_parts.append(c.text)

        unsub = self.agent.subscribe(handler)
        try:
            await self.agent.prompt(text)
            response = "".join(response_parts)
        finally:
            unsub()

        await self.state.set("last_messages", [m.dict() for m in self.agent.state.messages])
        return response

    async def on_wake(self):
        """Reconnect resources after sleep."""
        if not self.mcp_client:
            self.mcp_client = ShopifyMCPClient(...)
            await self.mcp_client.initialize()

    async def on_terminate(self):
        """Final cleanup."""
        if self.mcp_client:
            await self.mcp_client.close()

The Gateway Is 6 Lines

@app.post("/webhook")
async def webhook(request: Request, background_tasks: BackgroundTasks):
    body = await request.json()
    for msg in extract_whatsapp_messages(body):
        actor = await rivet.get_actor(ShoppingAgentActor, msg["sender"])
        background_tasks.add_task(actor.on_message, msg["text"])
    return {"status": "ok"}

That’s the entire FastAPI gateway. Everything else lives in the actor.

Scaling

Because actors are distributed by Rivet:

graph TB
    LB[Load Balancer] --> G1[Gateway Instance 1]
    LB --> G2[Gateway Instance 2]
    LB --> G3[Gateway Instance 3]

    G1 --> RV[Rivet Cluster]
    G2 --> RV
    G3 --> RV

    RV --> A1[Actor: User A]
    RV --> A2[Actor: User B]
    RV --> A3[Actor: User C]
    RV --> AN[Actor: User N]

    style RV fill:#f3e5f5,stroke:#7b1fa2,stroke-width:3px

• Multiple gateway instances → no sticky sessions needed
• Actors distributed across Rivet nodes
• Any gateway can route to any actor
• Cluster scales horizontally

Comparison: Old vs New Architecture

See Also

• Rivet Actor Model — Full implementation with complete code
• Agent Architecture — Detailed agent stack
• Architecture — High-level overview