Build a Production-Ready Gmail Agent with NSP Python SDK

NSP attaches to Gmail running in Chrome and exposes a live state tree you can query and act on from Python. This guide walks you through every step — from connecting to the Chrome debug port all the way to streaming new-email notifications as they land.

Prerequisites

Before you write any code, make sure the following are in place:

Chrome Debug Port

Chrome must be running with --remote-debugging-port=9222. See the Chrome Setup guide for launch instructions.

Gmail Open

Open mail.google.com in a Chrome tab. The daemon matches the tab by URL within about 5 seconds.

NSP Daemon Running

Run axon-daemon.exe run (or start the Windows Service). The daemon listens on http://localhost:7842 by default.

SDK Installed

Install the Python SDK: pip install nelieo-nsp

Step 1 — Verify Gmail Is Detected

Before building your agent, confirm the daemon has picked up Gmail. Run a quick list check:

import asyncio
import nelieo_nsp as nsp

async def main():
    async with nsp.NSPClient(api_key="sk_nsp_your_key_here") as client:
        apps = await client.list_apps()
        for app in apps:
            print(f"{app.app_name} | conf={app.confidence:.2f} | pid={app.pid}")

asyncio.run(main())
# Gmail | conf=0.97 | pid=12847

Gmail should appear with confidence >= 0.95. If it doesn’t show up yet, wait 30–60 seconds for the cold probe to complete.

Step 2 — Connect to Gmail

Use wait_for_app instead of a bare attach call. It retries until Gmail is ready, which is safer on startup:

import asyncio
import nelieo_nsp as nsp
from loguru import logger

API_KEY = "sk_nsp_your_key_here"

async def connect():
    client = nsp.NSPClient(api_key=API_KEY)
    await client.__aenter__()

    logger.info("Waiting for Gmail...")
    session = await client.wait_for_app("Gmail", timeout=30.0)

    logger.info(
        "Connected to {} (conf={:.2f}, {} actions available)",
        session.app_name,
        session.confidence,
        len(session.actions),
    )
    return client, session

Step 3 — Read Inbox State

Call session.refresh() before reading to get the latest snapshot, then pull fields from the state tree using typed accessors:

async def get_inbox_summary(session) -> dict:
    """Return unread count and a list of email metadata dicts."""
    await session.refresh()
    state = session.state

    unread = state.get_int("inbox.unread_count")
    total  = state.get_int("inbox.total_count")

    emails = []
    for i in range(min(unread, 20)):
        prefix = f"inbox.emails[{i}]"
        email_id = state.get_str(f"{prefix}.id")
        if not email_id:
            break
        emails.append({
            "id":              email_id,
            "thread_id":       state.get_str(f"{prefix}.thread_id"),
            "from":            state.get_str(f"{prefix}.from"),
            "subject":         state.get_str(f"{prefix}.subject"),
            "snippet":         state.get_str(f"{prefix}.snippet"),
            "is_read":         state.get_bool(f"{prefix}.is_read"),
            "has_attachments": state.get_bool(f"{prefix}.has_attachments"),
        })

    return {"unread_count": unread, "total_count": total, "emails": emails}

Available inbox state keys

Key	Type	Description
`inbox.unread_count`	`int`	Number of unread messages
`inbox.total_count`	`int`	Total messages in inbox
`inbox.sent_count`	`int`	Messages in Sent folder
`inbox.emails[N].id`	`str`	Unique message ID
`inbox.emails[N].thread_id`	`str`	Thread the message belongs to
`inbox.emails[N].from`	`str`	Sender address
`inbox.emails[N].subject`	`str`	Message subject line
`inbox.emails[N].snippet`	`str`	Short preview of the body
`inbox.emails[N].is_read`	`bool`	Whether the message is read
`inbox.emails[N].has_attachments`	`bool`	Whether attachments are present

Step 4 — Send a Reply

send_reply and send_email are both irreversible_write actions. Once dispatched, the message is sent and cannot be recalled. Always pass verify=True with a verify_expression to confirm the send succeeded before your agent continues.

async def send_reply(session, thread_id: str, body: str, cc: list[str] = None) -> bool:
    """Send a reply to a thread and verify it was delivered."""
    from loguru import logger

    # Capture the current sent count so we can verify it increments
    before_sent = session.get_int("inbox.sent_count")

    result = await session.execute(
        "send_reply",
        parameters={
            "thread_id": thread_id,
            "body": body,
            **( {"cc": cc} if cc else {} ),
        },
        verify=True,
        verify_expression=f"inbox.sent_count > {before_sent}",
        verify_timeout_ms=8000,
    )

    if result.success:
        logger.info("Reply sent in {}ms", result.latency_ms)
    else:
        logger.error("Reply failed: {}", result.error_message)

    return result.success

Step 5 — Archive and Mark as Read

These are reversible_write actions so they don’t require a verify_expression, though adding one is still good practice:

async def archive(session, email_id: str) -> bool:
    result = await session.execute(
        "archive",
        parameters={"email_id": email_id},
    )
    return result.success

async def mark_read(session, email_id: str) -> bool:
    result = await session.execute(
        "mark_read",
        parameters={"email_id": email_id},
    )
    return result.success

Step 6 — Stream New Emails in Real Time

Use session.watch() with auto_refresh_on_event=True so the session state is always current inside the loop. Filter on inbox.unread_count to detect arrivals:

async def watch_inbox(session, handler):
    """Call handler(email_dict) whenever a new email arrives at the top."""
    last_unread = session.get_int("inbox.unread_count")
    last_top_id = session.get_str("inbox.emails[0].id")

    async with session.watch(auto_refresh_on_event=True) as stream:
        async for event in stream:
            if "inbox.unread_count" not in event.changed_keys:
                continue

            new_unread = session.get_int("inbox.unread_count")
            new_top_id = session.get_str("inbox.emails[0].id")

            if new_top_id != last_top_id and new_unread > last_unread:
                email = {
                    "id":      new_top_id,
                    "from":    session.get_str("inbox.emails[0].from"),
                    "subject": session.get_str("inbox.emails[0].subject"),
                }
                await handler(email)

            last_unread = new_unread
            last_top_id = new_top_id

The /watch WebSocket endpoint is not subject to the daemon’s rate limiter. You can keep a stream open indefinitely without consuming your request quota.

Complete Agent

Here is the full working agent combining all of the above steps:

import asyncio
import nelieo_nsp as nsp
from loguru import logger

API_KEY = "sk_nsp_your_key_here"


class GmailAgent:
    def __init__(self):
        self.client = nsp.NSPClient(api_key=API_KEY)
        self.session = None

    async def connect(self):
        await self.client.__aenter__()
        logger.info("Waiting for Gmail...")
        self.session = await self.client.wait_for_app("Gmail", timeout=30.0)
        logger.info(
            "Connected to {} (conf={:.2f}, {} actions)",
            self.session.app_name,
            self.session.confidence,
            len(self.session.actions),
        )

    async def disconnect(self):
        await self.client.aclose()

    # ── Reading state ──────────────────────────────────────────────────────

    async def get_inbox_summary(self) -> dict:
        await self.session.refresh()
        state = self.session.state

        unread = state.get_int("inbox.unread_count")
        total  = state.get_int("inbox.total_count")

        emails = []
        for i in range(min(unread, 20)):
            prefix   = f"inbox.emails[{i}]"
            email_id = state.get_str(f"{prefix}.id")
            if not email_id:
                break
            emails.append({
                "id":              email_id,
                "thread_id":       state.get_str(f"{prefix}.thread_id"),
                "from":            state.get_str(f"{prefix}.from"),
                "subject":         state.get_str(f"{prefix}.subject"),
                "snippet":         state.get_str(f"{prefix}.snippet"),
                "is_read":         state.get_bool(f"{prefix}.is_read"),
                "has_attachments": state.get_bool(f"{prefix}.has_attachments"),
            })

        return {"unread_count": unread, "total_count": total, "emails": emails}

    # ── Actions ────────────────────────────────────────────────────────────

    async def send_reply(self, thread_id: str, body: str, cc: list[str] = None) -> bool:
        before_sent = self.session.get_int("inbox.sent_count")

        result = await self.session.execute(
            "send_reply",
            parameters={
                "thread_id": thread_id,
                "body": body,
                **( {"cc": cc} if cc else {} ),
            },
            verify=True,
            verify_expression=f"inbox.sent_count > {before_sent}",
            verify_timeout_ms=8000,
        )

        if result.success:
            logger.info("Reply sent in {}ms", result.latency_ms)
        else:
            logger.error("Reply failed: {}", result.error_message)
        return result.success

    async def archive(self, email_id: str) -> bool:
        result = await self.session.execute(
            "archive",
            parameters={"email_id": email_id},
        )
        return result.success

    async def mark_read(self, email_id: str) -> bool:
        result = await self.session.execute(
            "mark_read",
            parameters={"email_id": email_id},
        )
        return result.success

    # ── Streaming ──────────────────────────────────────────────────────────

    async def watch_inbox(self, handler):
        last_unread = self.session.get_int("inbox.unread_count")
        last_top_id = self.session.get_str("inbox.emails[0].id")

        async with self.session.watch(auto_refresh_on_event=True) as stream:
            async for event in stream:
                if "inbox.unread_count" not in event.changed_keys:
                    continue

                new_unread = self.session.get_int("inbox.unread_count")
                new_top_id = self.session.get_str("inbox.emails[0].id")

                if new_top_id != last_top_id and new_unread > last_unread:
                    email = {
                        "id":      new_top_id,
                        "from":    self.session.get_str("inbox.emails[0].from"),
                        "subject": self.session.get_str("inbox.emails[0].subject"),
                    }
                    logger.info("New email from: {}", email["from"])
                    await handler(email)

                last_unread = new_unread
                last_top_id = new_top_id


# ── Entry point ────────────────────────────────────────────────────────────────

async def main():
    agent = GmailAgent()
    await agent.connect()

    # Print inbox summary
    inbox = await agent.get_inbox_summary()
    print(f"\nInbox: {inbox['unread_count']} unread / {inbox['total_count']} total\n")
    print(f"{'':2}{'From':<30} {'Subject':<50} Read")
    print("─" * 90)
    for email in inbox["emails"]:
        status = "✓" if email["is_read"] else "●"
        print(f"{status}  {email['from'][:28]:<30} {email['subject'][:48]:<50}")

    # Watch for new emails and auto-reply
    print("\nWatching for new emails (Ctrl+C to stop)...")

    async def on_new_email(email):
        print(f"\n📧 New email from {email['from']}: {email['subject']}")
        inbox = await agent.get_inbox_summary()
        if inbox["emails"]:
            thread_id = inbox["emails"][0]["thread_id"]
            await agent.send_reply(
                thread_id,
                "Thank you for your message. I'll review this shortly.",
            )

    await agent.watch_inbox(on_new_email)


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

Available Gmail Actions

Action	Reversibility	Required Parameters	Optional
`send_reply`	`irreversible_write`	`thread_id`, `body`	`cc`
`send_email`	`irreversible_write`	`to`, `subject`, `body`	`cc`
`archive`	`reversible_write`	`email_id`	—
`mark_read`	`reversible_write`	`email_id`	—
`mark_unread`	`reversible_write`	`email_id`	—
`star`	`reversible_write`	`email_id`	—
`delete`	`irreversible_write`	`email_id`	—
`label`	`reversible_write`	`email_id`, `label`	—
`search`	`read`	`query`	—
`get_thread`	`read`	`thread_id`	—

Start with reversible_write actions like archive and mark_read while testing. Only enable send_reply and send_email once you’re confident in your agent’s logic.

Get Started

The Daemon

Runtime Probes

Security

Guides

Prerequisites

Chrome Debug Port

Gmail Open

NSP Daemon Running

SDK Installed

Step 1 — Verify Gmail Is Detected

Step 2 — Connect to Gmail

Step 3 — Read Inbox State

Available inbox state keys

Step 4 — Send a Reply

Step 5 — Archive and Mark as Read

Step 6 — Stream New Emails in Real Time

Complete Agent

Available Gmail Actions

​Prerequisites

Chrome Debug Port

Gmail Open

NSP Daemon Running

SDK Installed

​Step 1 — Verify Gmail Is Detected

​Step 2 — Connect to Gmail

​Step 3 — Read Inbox State

​Available inbox state keys

​Step 4 — Send a Reply

​Step 5 — Archive and Mark as Read

​Step 6 — Stream New Emails in Real Time

​Complete Agent

​Available Gmail Actions

Prerequisites

Step 1 — Verify Gmail Is Detected

Step 2 — Connect to Gmail

Step 3 — Read Inbox State

Available inbox state keys

Step 4 — Send a Reply

Step 5 — Archive and Mark as Read

Step 6 — Stream New Emails in Real Time

Complete Agent

Available Gmail Actions