By the end of this guide you’ll have a working support agent. It listens for email arriving at your Nevo address, hands each message to a language model, and replies on the same thread — roughly 40 lines of Python, no framework.
We’ll use Claude for the model. Any LLM with a similar API works — the Nevo side is identical.
What you’re building
┌──────────────────────────┐
│ Customer sends email to │
│ your support address │
└────────────┬─────────────┘
▼
┌──────────────────────────┐
│ Auto-forwarded to Nevo │
│ (email.received) │
└────────────┬─────────────┘
▼
┌──────────────────────────┐
│ Your agent │
│ ─ reads event.prompt_ready
│ ─ asks Claude for reply │
│ ─ event.reply(text=...) │
└────────────┬─────────────┘
▼
┌──────────────────────────┐
│ Reply lands in the │
│ sender's inbox, │
│ threaded on the same │
│ conversation │
└──────────────────────────┘
Prerequisites
- Python 3.10+
- A Nevo project with an email channel configured — see Create a project
- An Anthropic API key (or swap for OpenAI, Gemini, etc.)
- Two environment variables set:
export NEVO_API_KEY=nvo_live_... export ANTHROPIC_API_KEY=sk-ant-...
1. Install the SDKs
pip install nevo-sdk anthropic
2. The skeleton
Start with a handler that just prints incoming events. This verifies the inbound side works before adding the model.
# agent.py
import asyncio, os
from nevo import Nevo
async def main():
async with Nevo(token=os.environ["NEVO_API_KEY"]) as client:
@client.on_event()
async def handle(event):
print(event.type, event.id)
await client.run()
asyncio.run(main())
Run it:
python agent.py
Send a test email to your Nevo address (<channel-id>@in.nevo.sh) from any mailbox you control. Within a second you should see:
email.received evt_01hk8def…
If nothing shows up, check your channel is active in the dashboard and verify you’re sending to the right address.
3. Add the language model
Now we call Claude with each event. event.prompt_ready is the key move here — Nevo has already rendered the email into clean text you can feed directly into the system prompt. No provider-specific parsing.
import asyncio, os
from anthropic import AsyncAnthropic
from nevo import Nevo
claude = AsyncAnthropic() # reads ANTHROPIC_API_KEY from env
SYSTEM = """You are a support agent for Acme Co. You receive an incoming
message that's already been rendered as readable text. Reply warmly and
concisely — no more than three short paragraphs. If the customer's
request needs escalation, say so plainly and explain the next step.
Never promise a specific deadline."""
async def draft_reply(event_text: str) -> str:
msg = await claude.messages.create(
model="claude-sonnet-4-6",
max_tokens=512,
system=SYSTEM,
messages=[{"role": "user", "content": event_text}],
)
# The content is a list of blocks; for a text response there's exactly one.
return msg.content[0].text
async def main():
async with Nevo(token=os.environ["NEVO_API_KEY"]) as client:
@client.on_event()
async def handle(event):
if event.type != "email.received":
return
reply = await draft_reply(event.prompt_ready)
print(f"[draft] {reply[:120]}…")
await client.run()
asyncio.run(main())
Run it. Send another test email. You should see the drafted reply print before it goes out.
4. Send the reply
Swap print for event.reply(). Nevo handles the email threading (RFC 5322 In-Reply-To + References) so the response lands in the same conversation in Gmail, Outlook, and Apple Mail.
@client.on_event()
async def handle(event):
if event.type != "email.received":
return
reply = await draft_reply(event.prompt_ready)
await event.reply(text=reply)
print(f"replied to {event.email.from_} — {event.id}")
That’s the whole thing. An email arrives at your Nevo address, the agent reads it, Claude drafts a reply, and the reply lands back in the sender’s inbox threaded on the original conversation.
5. The full file
# agent.py
import asyncio, os
from anthropic import AsyncAnthropic
from nevo import Nevo
claude = AsyncAnthropic()
SYSTEM = """You are a support agent for Acme Co. You receive an incoming
message that's already been rendered as readable text. Reply warmly and
concisely — no more than three short paragraphs. If the customer's
request needs escalation, say so plainly and explain the next step.
Never promise a specific deadline."""
async def draft_reply(event_text: str) -> str:
msg = await claude.messages.create(
model="claude-sonnet-4-6",
max_tokens=512,
system=SYSTEM,
messages=[{"role": "user", "content": event_text}],
)
return msg.content[0].text
async def main():
async with Nevo(token=os.environ["NEVO_API_KEY"]) as client:
@client.on_event()
async def handle(event):
if event.type != "email.received":
return
reply = await draft_reply(event.prompt_ready)
await event.reply(text=reply)
print(f"replied to {event.email.from_} — {event.id}")
await client.run()
asyncio.run(main())
Under 30 lines of your own code, no framework, one model call per message.
What’s next
A few directions for a real deployment:
- Point real customer email at the channel. Up to now you’ve been sending to
<channel-id>@in.nevo.shdirectly for testing. In production, put that address behind your existing support inbox — e.g. add an auto-forwarding rule in Gmail or Google Workspace fromsupport@yourcompany.comto the Nevo address. The original sender’sFrom:is preserved, so whenevent.reply()fires it lands in the customer’s inbox, not yours. Avoid the manual “click forward” button in a mail client — that rewritesFrom:to you, and the agent would reply to you instead of the customer. - Guard against replying to your own bot. If customers reply to the bot’s reply, you’ll loop. Skip messages from your own inbound address or where the subject already starts with
Re:and the thread is deep. - Use
event.originto distinguish live events from dashboard-triggered replays. Your agent probably wants to skip replays to avoid duplicate work. - Escalate on uncertainty. Add a second model call that classifies the request — bill it as “answerable”, “needs human”, or “spam” — and only auto-reply on “answerable”.
- Log the full thread. Every event is persisted by Nevo for the replay window of your plan. The dashboard shows the agent’s replies alongside the incoming events, so you can audit what it said.
Swapping the model
The Nevo half of this guide is model-agnostic. To use OpenAI, replace the Anthropic import + call:
from openai import AsyncOpenAI
openai = AsyncOpenAI()
async def draft_reply(event_text: str) -> str:
resp = await openai.chat.completions.create(
model="gpt-5",
messages=[
{"role": "system", "content": SYSTEM},
{"role": "user", "content": event_text},
],
)
return resp.choices[0].message.content
Same shape for Google Gemini, Mistral, local llama.cpp, etc. The handler itself doesn’t change.
Related
- Events — the full event shape your handler receives.
- Prompt-ready — how the rendering works, and when to build your own prompt instead.
- Python SDK: replies — every option
event.reply()accepts.