Metadata-Version: 2.4
Name: pipbit
Version: 0.8.0
Summary: Python SDK for connecting text-based agents to the Pipbit platform.
Author-email: Pipbit <pipbit@pipworks.ai>
License-Expression: MIT
Project-URL: Homepage, https://pipworks.ai/developer
Project-URL: Documentation, https://pipworks.ai/developer/docs/sdk
Keywords: pipbit,ai,agent,sdk,flask
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Framework :: Flask
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.9
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: Flask<4.0,>=3.0
Requires-Dist: requests<3.0,>=2.31
Provides-Extra: dev
Requires-Dist: build<2.0,>=1.2; extra == "dev"
Requires-Dist: twine<7.0,>=6.2; extra == "dev"
Dynamic: license-file

# Pipbit Python SDK

`pipbit` is the Python SDK for building text-based developer agents that Pip can hand conversations to.

Pipbit keeps the voice stack on the platform side:
- microphone and speaker handling
- speech-to-text and text-to-speech
- device transport
- request signing
- handoff session management

Your agent stays text-first and only needs to implement a webhook handler.

## Installation

```bash
pip install pipbit
```

For local package work:

```bash
pip install -e '.[dev]'
```

Upgrading an agent in place: use a versioned upgrade for released use
(`pip install -U 'pipbit==0.8.0'`), or an editable install of the SDK checkout
when dogfooding from the monorepo (`pip install -e path/to/SDK/pipbit_sdk` —
so the venv tracks the source instead of freezing a stale copy). Either way,
**restart the serving process after upgrading**: a long-lived worker keeps the
old module in memory, and agent code written against a newer SDK will crash
against the stale import.

## Hello world

```python
from pipbit import Agent, CONNECT_INTRO_UTTERANCE

agent = Agent(
    name="george",
    secret="pb_live_sk_XXXXXXXXXXXXXXXX",
    base_url="https://api.pipbit.ai",
)

@agent.handle
def handle(req):
    if req.is_connect_intro():
        return "George here. How can I help with your home DIY today?"
    return f"You said: {req.text}"

agent.run(host="0.0.0.0", port=3000)
```

Pipbit sends signed JSON requests to:

```text
POST /pipbit/agent
```

`CONNECT_INTRO_UTTERANCE` is exported for agent authors who want the raw token,
but `req.is_connect_intro()` is the preferred helper.

`Agent(name=...)` must be your **lowercase canonical handle** from
registration ("George" registers as `george`): every webhook carries it in
`X-Pipbit-Agent`, and the SDK requires a byte-identical match — a case
mismatch fails authentication on every turn.

### Environment variables

`Agent.from_env()` builds an agent from the standard variables —
`PIPBIT_AGENT_NAME`, `PIPBIT_AGENT_SECRET`, and optionally `PIPBIT_BASE_URL`
and `PIPBIT_API_VERSION`. Keyword arguments override the environment and pass
through to `Agent()` (e.g. `oauth=`, `token_store=`):

```python
agent = Agent.from_env()
```

The SDK does not load a `.env` file itself — pair with `python-dotenv` if you
keep the variables in one.

## Webhook protocol

Each request carries three headers:

```text
X-Pipbit-Agent:      george            <- your canonical handle
X-Pipbit-Timestamp:  1753212345        <- unix seconds
X-Pipbit-Signature:  v1=<hex digest>
```

The signature is HMAC-SHA256, keyed by your agent secret, over the exact
bytes `"<timestamp>." + raw_request_body`, hex-encoded. The SDK verifies the
name, the timestamp (a **60-second replay window** — keep your clock synced),
and the signature before your handler runs, answering 401 otherwise.

To hand-build a signed request in tests, the SDK exports its signing helper
and header constants:

```python
import json, time
from pipbit import (
    build_signature,
    PIPBIT_AGENT_HEADER, PIPBIT_TIMESTAMP_HEADER, PIPBIT_SIGNATURE_HEADER,
)

body = json.dumps({
    "request_id": "req_1", "reply_token": "pb_rt_1",
    "utterance": "hello", "subject_id": "pb_sub_1", "conversation": [],
}).encode("utf-8")
ts = str(int(time.time()))
headers = {
    "Content-Type": "application/json",
    PIPBIT_AGENT_HEADER: "george",
    PIPBIT_TIMESTAMP_HEADER: ts,
    PIPBIT_SIGNATURE_HEADER: "v1=" + build_signature("pb_live_sk_test", ts, body),
}
```

## Request model

Each inbound request is exposed as a `Request` object:

```python
req.id                  # wire field: request_id
req.reply_token         # one-shot token for push_reply
req.session_id
req.text                # wire field: utterance
req.conversation
req.subject_id
req.agent_session_id
req.device_id
req.device_shadow
req.handoff_context
req.environment_event
req.manager_name        # e.g. "Pippa"; None on older platforms
req.is_connect_intro()
req.is_environment_event()
req.defer()
```

- `subject_id` is the stable Pipbit subject id for the current user/device binding.
- `session_id` is the request/reply conversation id.
- `agent_session_id` is the higher-level Pip-managed handoff session.
- `manager_name` is the user's chosen name for their manager assistant. Use it
  in send-offs instead of "the manager".
- `conversation` carries the last few turns of the routed conversation (user
  turns, and on current platforms your agent's spoken replies too). It is
  short and resets when a new agent session starts — agents that reason over
  context should keep their own per-`subject_id` history of what they said.

## Reply modes

Immediate reply:

```python
@agent.handle
def handle(req):
    return "Hi. How can I help?"
```

Deferred reply:

```python
@agent.handle
def handle(req):
    queue_work(req)
    return req.defer()
```

Later:

```python
agent.push_reply(
    reply_token=req.reply_token,
    text="Your report is ready.",
)
```

Reply tokens are **one-shot** and expire about 15 minutes after the turn. A
second `push_reply` on the same token is rejected — which also makes a
network-error retry safe: if the first attempt landed, the retry fails with
"not pending" instead of speaking twice. A `202` means the platform owns
delivery from there (it retries transient device conditions itself and holds
the reply for a quiet moment); do not re-send after a `202`.

If the reply arrives after the conversational moment has passed, `push_reply`
raises `SessionGoneError` — see [Error handling](#error-handling).

Handing back (ending the agent's turn):

```python
agent.push_reply(
    reply_token=req.reply_token,
    text=send_off,       # your LLM's own goodbye, naming req.manager_name
    handback=True,
)
```

With `handback=True` this reply is the agent's send-off: after it is spoken,
the platform returns control to the manager assistant automatically. The
platform adds no farewell on your behalf — say the goodbye yourself (the
manager then acknowledges her return in her own voice). Let your LLM word it
for the moment (Garry's `handback_to_manager` tool is the worked pattern) and
name the manager via `req.manager_name`. The flag is omitted from the payload
when `False`, so older platforms are unaffected.

Proactive message:

```python
agent.push_message(
    subject_id=req.subject_id,
    text="You have a new update.",
)
```

A pushed message is spoken right away when the user has a live session; when
the device is busy the platform retries for about a minute and can hold it
for the next quiet moment. If the user is not around it lands in the device
inbox rather than being spoken later — treat `push_message` as "say this now
if possible". There is no idempotency key, so do not blind-retry it.

## Environment events

Agents can subscribe to real-time environment events (audio scene, sensor
data, etc.) by including `event_subscriptions` during registration:

```json
{"event_subscriptions": ["audio_scene"]}
```

Events arrive as normal webhook calls with a sentinel utterance:

```python
@agent.handle
def handle(req):
    if req.is_environment_event():
        event = req.environment_event  # {"type": "audio_scene", "data": {...}, "timestamp": ...}
        if req.agent_session_id:
            return "I heard something."       # active session — reply directly
        else:
            agent.push_message(               # background — signal Pip
                text="Alert from my agent.",
                device_id=req.device_id,
            )
            return req.defer()   # no spoken reply for this event (an empty string is rejected)
    # ... normal handler
```

Additional request fields for environment events:

- `req.environment_event` — structured event dict (`type`, `data`, `timestamp`)
- `req.device_id` — the device identifier (always present for events)
- `req.is_environment_event()` — returns `True` for event requests

`ENVIRONMENT_EVENT_UTTERANCE` is exported for agents that want the raw
sentinel token.

## OAuth account-linking

Connect a user's third-party account (Sonos, Spotify, Google, ...) to your
agent. **Your agent owns the tokens — the platform never sees them.** Because a
Pipbit device is voice-only, the user can't consent on the device; instead the
agent hands off to a second screen (the same way Alexa/Google "account linking"
works): you generate a link, deliver it to the user out of band, they consent in
a browser, and the SDK's built-in callback stores the token under their
`subject_id`.

```python
from pipbit import Agent, OAuth2Config, FileTokenStore

agent = Agent(
    name="sonos",
    secret="pb_live_sk_...",
    oauth=OAuth2Config(
        authorize_url="https://api.sonos.com/login/v3/oauth",
        token_url="https://api.sonos.com/login/v3/oauth/access",
        client_id="...",
        client_secret="...",
        redirect_uri="https://my-agent.example.com/pipbit/oauth/callback",
        scope="playback-control-all",
        token_endpoint_auth="basic",   # or "post"
    ),
    token_store=FileTokenStore("tokens.json"),   # MemoryTokenStore() for dev
)

@agent.handle
def handle(req):
    token = agent.get_oauth_token(req.subject_id)   # auto-refreshes; None if unlinked
    if token is None:
        link = agent.account_link_url(req.subject_id)
        deliver_to_user(link)        # your channel: SMS / email / companion app
        return "I've sent you a link to connect your account — open it, sign in, and tap allow."
    return do_something(token.access_token)
```

When `oauth=` is set, the SDK automatically serves the callback:

```text
GET /pipbit/oauth/callback
```

so your `redirect_uri` must point there (and HTTPS). The `state` parameter is
HMAC-signed with your agent secret and time-limited, so a third party can't
forge a callback to link the wrong identity.

Helpers:

- `agent.account_link_url(subject_id)` — the URL to deliver to the user.
- `agent.get_oauth_token(subject_id)` — a valid `OAuth2Token` (refreshed if
  needed), or `None` if not linked / no longer refreshable.
- `agent.is_linked(subject_id)` / `agent.unlink(subject_id)`.
- `on_account_link=` (an `Agent` kwarg) fires `fn(subject_id, token)` after a
  successful link — handy to `push_message` the user a confirmation.

### Voice-first onboarding (`start_device_link`)

`account_link_url()` gives you the raw provider URL, but a screenless device
can't hand the user a link. For the voice path, use `start_device_link()` — the
platform brokers a fixed verification URL (and a spoken code when the user has no
phone on file), so you just tell the user where to go:

```python
link = agent.start_device_link(req.subject_id, provider_label="Sonos")
if link.method == "code":
    return f"Go to {link.verification_uri} and enter {link.user_code} to connect your Sonos."
return f"Open {link.verification_uri} to finish connecting your Sonos."
```

The user finishes consent there; the token still lands at this agent's
`/pipbit/oauth/callback`, and `get_oauth_token()` returns it next turn.

> `start_device_link()` uses the platform's `POST /v1/account-links/start`
> broker, which is live — it is the recommended path for voice-first linking.
> `account_link_url()` remains the manual alternative when you deliver the
> provider URL yourself (a spoken URL is useless, so use a code/second-screen
> flow rather than an SMS link).

## Health checks

The SDK exposes:

- `GET /health`
- `GET /pipbit/health`

The response is structured and readiness-aware. The SDK includes:

- a required `agent_secret` check
- a non-required `platform_health` check against `GET {base_url}/health`

Add your own checks for agent-specific dependencies:

```python
import os

from pipbit import Agent

agent = Agent(name="george", secret="pb_live_sk_...", base_url="https://api.pipbit.ai")

agent.add_health_check(
    "openai_api_key",
    lambda: (bool(os.getenv("OPENAI_API_KEY")), "Set OPENAI_API_KEY."),
    required=True,
)
```

Example health response:

```json
{
  "status": "ok",
  "ready": true,
  "agent_name": "george",
  "api_version": "v1",
  "base_url": "https://api.pipbit.ai",
  "checks": [
    {"name": "agent_secret", "ok": true, "required": true, "message": "ok"},
    {"name": "platform_health", "ok": true, "required": false, "message": "ok"}
  ]
}
```

If any required check fails, `/health` returns `503`.

## Metrics

The SDK exposes:

- `GET /metrics`
- `GET /pipbit/metrics`

Current counters include:

- `webhook_requests_total`
- `webhook_success_total`
- `webhook_immediate_replies_total`
- `webhook_deferred_replies_total`
- `webhook_auth_failures_total`
- `webhook_invalid_requests_total`
- `webhook_configuration_errors_total`
- `webhook_handler_errors_total`
- `push_reply_calls_total`
- `push_message_calls_total`
- `device_link_requests_total`
- `api_request_errors_total`
- `oauth_callback_requests_total`
- `platform_health_checks_total`
- `platform_health_check_errors_total`

## Local development

Run your Flask app locally:

```bash
python app.py
```

For local testing, expose your server so Pipbit can reach `POST /pipbit/agent`
(an HTTPS tunnel such as cloudflared or ngrok works well).

## Production serving

`agent.app()` returns a standard Flask WSGI app. The shipped examples serve
it with gunicorn:

```python
# my_agent.py
app = agent.app()
```

```bash
gunicorn my_agent:app --bind 0.0.0.0:3000 --workers 1 --threads 8
```

Use **exactly one worker**: in-process state (conversation history, pending
link confirmations, `MemoryTokenStore` tokens) is per-process, and extra
workers shard it — the agent gets amnesia between turns. One worker is enough
because handlers defer instantly and do real work on background threads;
`--threads 8` keeps a slow in-request path (like the OAuth callback's token
exchange) from blocking the webhook. To scale past one process, move per-user
state to shared storage first. Use `/health` for load-balancer readiness — it
returns 503 until required checks pass.

The spoken voice is configured on the Pipbit platform record for your agent; it
is not chosen inside the Python SDK. The preferred platform contract is:

- `voice_request`: developer-supplied abstract preference like presentation, tone, role
- `voice_binding`: platform-selected locked concrete voice
- `voice_id`: bridge-compatible compatibility field derived from the binding

Legacy explicit `voice_id` registration still works, but new agents should
prefer `voice_request`.

## Contract test

This repo includes an SDK contract test that exercises the basic developer flow:

- register agent
- bind device/subject
- connect the agent
- send the handoff intro turn
- send one normal user turn
- end the agent session

Run it from the repo root:

```bash
PYTHONPATH=SDK/pipbit_sdk:HOST/pipbit_platform_service \
python3 -m unittest SDK.pipbit_sdk.tests.test_contract -v
```

## Packaging for PyPI

Build the sdist and wheel:

```bash
cd SDK/pipbit_sdk
python3 -m build
twine check dist/*
```

The repo also includes a `RELEASE.md` checklist for version bumps and publish steps.

## Error handling

All SDK errors derive from `PipbitError`:

- `AuthenticationError` — a 401/403 response. Not always a bad secret: a
  suspended/unpublished agent or a reply token belonging to another agent
  403s too — the platform's detail rides the message and `.code`.
- `InvalidRequestError` — the platform rejected the request.
- `SessionGoneError` — subclass of `InvalidRequestError`: the reply token is
  unknown or expired, so the conversational moment is over. Fall back to
  `push_message` if the text is still worth saying:

  ```python
  from pipbit import SessionGoneError

  try:
      agent.push_reply(reply_token=token, text=answer)
  except SessionGoneError:
      agent.push_message(subject_id=subject_id, text=answer)
  ```

- `ConfigurationError` — the SDK is configured incorrectly.
- `OAuthError` — raised by the account-linking flow (`OAuth2Client.exchange_code()` /
  `refresh()`). `Agent.get_oauth_token()` and the built-in `/pipbit/oauth/callback`
  route handle it internally, so you only catch it if you call `OAuth2Client` directly.

Errors raised from platform responses carry `.code` (the platform's
machine-readable error string) and `.status_code`. Network failures and 5xx
responses raise plain `RuntimeError`.

Inbound webhook failures return JSON error bodies with sensible HTTP status codes.
