Build a Gmail Agent with NSP: Read, Reply, and Stream

This guide walks you through building a complete Gmail agent with NSP. By the end you will have a working Python script that reads your inbox, archives messages, sends verified replies, and reacts to new emails in real time — all by talking directly to Gmail’s live runtime state through the NSP daemon, with no browser automation or UI scraping involved.

Prerequisites

Before you start, confirm you have everything in place:

NSP daemon installed and running. Download the latest Nelieo NSP Setup.exe from platform.nelieo.com/download. The installer registers the daemon as a Windows Service that auto-starts on login and listens on port 7842.
Chrome launched with the remote debugging port enabled. NSP reads Gmail’s V8 state through the Chrome DevTools Protocol (CDP). You must start Chrome with the flag before opening Gmail.
Gmail open in a tab. Navigate to mail.google.com after Chrome is running.
Python 3.11+ and the NSP SDK installed.
An NSP API key from platform.nelieo.com.

Launch Chrome with CDP enabled:

"C:\Program Files\Google\Chrome\Application\chrome.exe" --remote-debugging-port=9222

Start-Process "chrome.exe" -ArgumentList "--remote-debugging-port=9222"

Close all existing Chrome windows before running either command. Chrome only supports one debugging session per user profile — if another instance is already running, it silently ignores the flag.

Install the SDK:

pip install nelieo-nsp

Set your API key as an environment variable so you never hardcode it:

# Windows (Command Prompt)
set NSP_API_KEY=nsp_live_sk_your_key_here

# Windows (PowerShell)
$env:NSP_API_KEY = "nsp_live_sk_your_key_here"

Connect to Gmail

NSPClient manages the connection to your local daemon. Call attach() with the app name — NSP resolves the correct process automatically, so you never need to track PIDs yourself.

import asyncio
import nelieo_nsp as nsp

async def main():
    async with nsp.NSPClient() as client:
        # Attach by name — no PID needed
        session = await client.attach("Gmail")
        print(f"Connected: {session.app_name}  confidence={session.confidence:.2f}")

asyncio.run(main())

Expected output:

Connected: Gmail  confidence=0.97

If Gmail has not been probed yet (first run), attach() may take up to 30 seconds while the daemon runs a cold V8 heap snapshot. On subsequent runs the schema is cached and attachment completes in under a second. Use client.wait_for_app("Gmail", timeout=30.0) if you want an explicit timeout with automatic retries.

Read Inbox State

Once attached, read typed state values directly from the session using dot-notation keys. NSP normalises Gmail’s V8 runtime data into the Substrate Schema Format (SSF) so every key is stable and predictable across probe cycles.

async with nsp.NSPClient() as client:
    session = await client.attach("Gmail")

    # Counts
    unread = session.get_int("inbox.unread_count")
    total  = session.get_int("inbox.total_count")
    print(f"You have {unread} unread emails ({total} total)")

    # First email metadata
    subject = session.get_str("inbox.emails[0].subject")
    sender  = session.get_str("inbox.emails[0].from")
    is_read = session.get_bool("inbox.emails[0].is_read")
    email_id   = session.get_str("inbox.emails[0].id")
    thread_id  = session.get_str("inbox.emails[0].thread_id")

    print(f"Latest: '{subject}' from {sender}  (read={is_read})")
    print(f"  email_id={email_id}  thread_id={thread_id}")

Expected output:

You have 47 unread emails (1284 total)
Latest: 'Q3 Performance Report' from alice@example.com  (read=False)
  email_id=msg_18821af3  thread_id=thread_7af2b3

List Available Actions

Every session exposes the actions that NSP discovered for the current app. Iterate session.actions to see what your agent can do, along with each action’s type signature and reversibility classification.

async with nsp.NSPClient() as client:
    session = await client.attach("Gmail")

    print(f"{'Action':<20} {'Reversibility':<22} Signature")
    print("-" * 80)
    for name, schema in session.actions.items():
        print(f"{name:<20} {schema.reversibility:<22} {schema.signature}")

Example output:

Action               Reversibility          Signature
--------------------------------------------------------------------------------
send_reply           irreversible_write     fn(thread_id: string, body: string, cc?: string[]) -> bool
send_email           irreversible_write     fn(to: string, subject: string, body: string, cc?: string[]) -> bool
archive              reversible_write       fn(email_id: string) -> bool
mark_read            reversible_write       fn(email_id: string) -> bool
mark_unread          reversible_write       fn(email_id: string) -> bool
star                 reversible_write       fn(email_id: string) -> bool
delete               irreversible_write     fn(email_id: string) -> bool
label                reversible_write       fn(email_id: string, label: string) -> bool
search               read                   fn(query: string) -> email[]
get_thread           read                   fn(thread_id: string) -> thread

Pay attention to the reversibility column. Actions classified as irreversible_write (like send_reply and delete) require you to pass verify=True and a verify_expression — NSP enforces this automatically and raises NSPIrreversibleNotConfirmedError if you omit it.

Archive an Email (Reversible)

archive is a reversible_write action — it moves an email out of your inbox but you can undo it by unarchiving later. Reversible actions do not require a verify_expression, though you can add one for extra confidence.

async with nsp.NSPClient() as client:
    session = await client.attach("Gmail")

    email_id = session.get_str("inbox.emails[0].id")
    print(f"Archiving {email_id}...")

    result = await session.execute(
        "archive",
        parameters={"email_id": email_id},
    )

    print(f"Archived: {result.success}  latency={result.latency_ms}ms")

Expected output:

Archiving msg_18821af3...
Archived: True  latency=148ms

Send a Reply (Irreversible)

send_reply is an irreversible_write action — once sent, the email cannot be recalled. NSP requires verify=True and a verify_expression for all irreversible actions. The daemon executes the action, then polls post-action state until your expression evaluates to true (or the timeout elapses).

async with nsp.NSPClient() as client:
    session = await client.attach("Gmail")

    thread_id   = session.get_str("inbox.emails[0].thread_id")
    before_sent = session.get_int("inbox.sent_count")

    result = await session.execute(
        "send_reply",
        parameters={
            "thread_id": thread_id,
            "body": "Thanks for the report — I'll review it today.",
            # "cc": ["manager@example.com"],  # optional
        },
        verify=True,
        verify_expression=f"inbox.sent_count > {before_sent}",
        verify_timeout_ms=8000,
    )

    print(f"Reply sent: {result.success}")
    print(f"Latency:    {result.latency_ms}ms")
    if result.verification:
        print(f"Verified:   {result.verification.passed}  "
              f"({result.verification.explanation})")

Expected output:

Reply sent: True
Latency:    312ms
Verified:   True  (inbox.sent_count changed from 14 to 15)

Omitting verify=True on an irreversible action raises NSPIrreversibleNotConfirmedError immediately — the action is never dispatched. This is a safety contract, not a hint.

Watch for New Emails (Real-Time Streaming)

Instead of polling for changes, open a WebSocket watch stream. The daemon fires an event on every probe cycle (roughly every 2 seconds) that contains a list of which state keys changed. Use auto_refresh_on_event=True so that session.get_*() calls inside the loop always return the latest values.

import asyncio
import nelieo_nsp as nsp

async def gmail_agent():
    async with nsp.NSPClient() as client:
        session = await client.attach("Gmail")
        print("Watching for new emails… (Ctrl+C to stop)")

        async with session.watch(auto_refresh_on_event=True) as stream:
            async for event in stream:
                # Only act when the unread count or top email changes
                if "inbox.unread_count" not in event.changed_keys:
                    continue

                unread  = session.get_int("inbox.unread_count")
                subject = session.get_str("inbox.emails[0].subject")
                sender  = session.get_str("inbox.emails[0].from")
                print(f"[seq={event.sequence}] {unread} unread — "
                      f"'{subject}' from {sender}")

asyncio.run(gmail_agent())

Expected output (as new mail arrives):

Watching for new emails… (Ctrl+C to stop)
[seq=1] 47 unread — 'Q3 Performance Report' from alice@example.com
[seq=4] 48 unread — 'Re: Budget Approval' from bob@example.com

Full Agent Script

The script below combines every step above into a single, copy-pasteable file. It reads the inbox on startup, prints a summary, then streams for new arrivals and auto-replies to each one.

import asyncio
import nelieo_nsp as nsp

async def gmail_agent():
    async with nsp.NSPClient() as client:
        # ── Connect ──────────────────────────────────────────────────────
        session = await client.attach("Gmail")
        print(f"Connected to {session.app_name}  "
              f"confidence={session.confidence:.2f}  "
              f"actions={len(session.actions)}")

        # ── Read inbox ───────────────────────────────────────────────────
        unread  = session.get_int("inbox.unread_count")
        subject = session.get_str("inbox.emails[0].subject")
        sender  = session.get_str("inbox.emails[0].from")
        print(f"\nYou have {unread} unread emails")
        print(f"Latest: '{subject}' from {sender}")

        # ── Archive the top email ────────────────────────────────────────
        email_id = session.get_str("inbox.emails[0].id")
        archive_result = await session.execute(
            "archive",
            parameters={"email_id": email_id},
        )
        print(f"\nArchived {email_id}: {archive_result.success}")

        # ── Send a reply to the second email ────────────────────────────
        thread_id   = session.get_str("inbox.emails[1].thread_id")
        before_sent = session.get_int("inbox.sent_count")

        reply_result = await session.execute(
            "send_reply",
            parameters={
                "thread_id": thread_id,
                "body": "Thanks, I'll review this shortly.",
            },
            verify=True,
            verify_expression=f"inbox.sent_count > {before_sent}",
            verify_timeout_ms=8000,
        )
        print(f"Reply sent: {reply_result.success}  "
              f"in {reply_result.latency_ms}ms")

        # ── Stream new emails ────────────────────────────────────────────
        print("\nWatching for new emails… (Ctrl+C to stop)")
        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_subject = session.get_str("inbox.emails[0].subject")
                new_sender  = session.get_str("inbox.emails[0].from")
                print(f"📧 New email: '{new_subject}' from {new_sender}")

asyncio.run(gmail_agent())

Expected output:

Connected to Gmail  confidence=0.97  actions=10

You have 47 unread emails
Latest: 'Q3 Performance Report' from alice@example.com

Archived msg_18821af3: True
Reply sent: True  in 312ms

Watching for new emails… (Ctrl+C to stop)
📧 New email: 'Re: Budget Approval' from bob@example.com

Get Started

The Daemon

Runtime Probes

Security

Guides

Full Agent Script

​Full Agent Script

Full Agent Script