crewai flows email checkpoint: adding human approval gates with lobstermail

pip install requests

# lobstermail_client.py
import time
import requests

LOBSTERMAIL_BASE = "https://api.lobstermail.ai/v1"

class LobsterMailClient:
    def __init__(self, token: str):
        self.headers = {"Authorization": f"Bearer {token}"}

    def create_inbox(self, name: str) -> dict:
        resp = requests.post(
            f"{LOBSTERMAIL_BASE}/inboxes",
            json={"name": name},
            headers=self.headers,
        )
        resp.raise_for_status()
        return resp.json()

    def send(self, from_addr: str, to: str, subject: str, body: str) -> None:
        resp = requests.post(
            f"{LOBSTERMAIL_BASE}/send",
            json={"from": from_addr, "to": to, "subject": subject, "body": body},
            headers=self.headers,
        )
        resp.raise_for_status()

    def poll_reply(self, inbox_id: str, timeout: int = 86400, interval: int = 30) -> str:
        """Poll until we get a reply. Returns 'approved', 'rejected', or 'timed_out'."""
        deadline = time.time() + timeout
        while time.time() < deadline:
            resp = requests.get(
                f"{LOBSTERMAIL_BASE}/inboxes/{inbox_id}/emails",
                headers=self.headers,
            )
            for email in resp.json().get("emails", []):
                body_lower = email.get("body", "").lower()
                if "approve" in body_lower:
                    return "approved"
                if "reject" in body_lower:
                    return "rejected"
            time.sleep(interval)
        return "timed_out"

# approval_flow.py
from crewai.flow.flow import Flow, listen, start, router
from crewai import Crew, Agent, Task
from lobstermail_client import LobsterMailClient

LOBSTERMAIL_TOKEN = "lm_sk_live_your_token_here"
APPROVER_EMAIL = "approver@yourcompany.com"

lm = LobsterMailClient(LOBSTERMAIL_TOKEN)


class ContentApprovalFlow(Flow):

    @start()
    def generate_draft(self):
        writer = Agent(
            role="Content Writer",
            goal="Write a compelling product announcement",
            backstory="Experienced writer specializing in developer tools",
        )
        task = Task(
            description="Write a 200-word product announcement for a new API feature",
            expected_output="A polished product announcement",
            agent=writer,
        )
        result = Crew(agents=[writer], tasks=[task]).kickoff()
        self.state["draft"] = str(result)
        return self.state["draft"]

    @router(generate_draft)
    def email_checkpoint(self, draft: str):
        # Fresh inbox per request — no cross-contamination from old replies
        inbox = lm.create_inbox("approval-bot")
        inbox_id = inbox["id"]
        inbox_addr = inbox["address"]

        body = f"""Your agent generated a draft for review.

--- DRAFT ---
{draft}
--- END DRAFT ---

Reply with APPROVE to publish, or REJECT: <reason> to send back for revision.
This inbox expires in 24 hours.
"""
        lm.send(inbox_addr, APPROVER_EMAIL, "Agent approval request", body)
        print(f"Approval request sent from {inbox_addr}")

        decision = lm.poll_reply(inbox_id)
        self.state["decision"] = decision
        return decision

    @listen("approved")
    def publish(self):
        print("Approved — publishing now")
        # publishing crew here

    @listen("rejected")
    def revise(self):
        print("Rejected — sending back for revision")
        self.generate_draft()  # restart the loop

    @listen("timed_out")
    def handle_timeout(self):
        print("No response in 24 hours — escalating")
        # escalation logic here


if __name__ == "__main__":
    ContentApprovalFlow().kickoff()