Python SDK

pip install capumattu

from capumattu import Capumattu

with Capumattu() as mr:  # reads MAILROOM_API_KEY and MAILROOM_URL
    inbox = mr.inboxes.create(
        username="support-bot",
        display_name="Support Bot",
        client_id="support-bot",  # idempotent
    )
    print(inbox["address"])  # support-bot@agents.capumattu.com

    # Wait for the next email (long-poll, up to 60 s). None on timeout.
    msg = mr.messages.wait(inbox["id"], timeout=60)
    if msg:
        print(msg["extracted_text"])  # the new part of the reply only
        mr.messages.reply(inbox["id"], msg["id"], text="Thanks, on it.")

import asyncio
from capumattu import AsyncCapumattu


async def main() -> None:
    async with AsyncCapumattu() as mr:
        inbox = await mr.inboxes.create()
        msg = await mr.messages.wait(inbox["id"], timeout=30)


asyncio.run(main())

c = mr.contacts.by_address("priya.n@gmail.com")
print(c["notes"], c["metadata"].get("crm_id"))

mr.contacts.update(
    c["id"],
    notes=(c["notes"] or "") + "
Asked about order 1042 on 25 Sep.",
    metadata={"crm_id": "hs_48213", "plan": "wholesale"},
    add_labels=["customer"],
)

mr.knowledge.create(
    title="Shipping times",
    body="- Germany: 1–2 working days (DHL)
- EU: 3–5 working days",
    inbox_id=inbox["id"],
)

hits = mr.knowledge.search("how long does shipping to France take", inbox_id=inbox["id"])
for h in hits["data"]:
    print(h["title"], h["rank"], h["snippet"])

draft = mr.messages.draft_reply(msg["id"], instructions="Keep it short.")
print(draft["text"], [c["title"] for c in draft["citations"]])
# review it, then:
mr.messages.reply(msg["inbox_id"], msg["id"], text=draft["text"])

from capumattu import iter_all

for msg in iter_all(mr.messages.list, inbox["id"], is_read=False):
    print(msg["subject"])