> ## Documentation Index
> Fetch the complete documentation index at: https://docs.nelieo.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Build a Production-Ready Gmail Agent with NSP Python SDK

> Build a Gmail agent with NSP: attach to Chrome, read inbox state, send verified replies, archive emails, and stream real-time new-email notifications.

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:

<CardGroup cols={2}>
  <Card title="Chrome Debug Port" icon="chrome">
    Chrome must be running with `--remote-debugging-port=9222`. See the [Chrome Setup guide](/guides/chrome-setup) for launch instructions.
  </Card>

  <Card title="Gmail Open" icon="envelope">
    Open `mail.google.com` in a Chrome tab. The daemon matches the tab by URL within about 5 seconds.
  </Card>

  <Card title="NSP Daemon Running" icon="server">
    Run `axon-daemon.exe run` (or start the Windows Service). The daemon listens on `http://localhost:7842` by default.
  </Card>

  <Card title="SDK Installed" icon="python">
    Install the Python SDK: `pip install nelieo-nsp`
  </Card>
</CardGroup>

## Step 1 — Verify Gmail Is Detected

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

```python theme={null}
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:

```python theme={null}
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:

```python theme={null}
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

<Warning>
  `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.
</Warning>

```python theme={null}
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:

```python theme={null}
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:

```python theme={null}
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
```

<Note>
  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.
</Note>

## Complete Agent

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

```python theme={null}
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`             | —        |

<Tip>
  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.
</Tip>