pipbit 0.7.0

Python SDK for connecting text-based agents to the Pipbit platform.

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

pip install pipbit

For local package work:

pip install -e '.[dev]'

Hello world

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:

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.

Request model

Each inbound request is exposed as a Request object:

req.id
req.reply_token
req.session_id
req.text
req.conversation
req.subject_id
req.agent_session_id
req.device_id
req.device_shadow
req.handoff_context
req.environment_event
req.is_connect_intro()
req.is_environment_event()

• 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.

Reply modes

Immediate reply:

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

Deferred reply:

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

Later:

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

Proactive message:

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

Environment events

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

{"event_subscriptions": ["audio_scene"]}

Events arrive as normal webhook calls with a sentinel utterance:

@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 ""
    # ... 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.

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:

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:

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() requires the platform's POST /v1/account-links/start broker. Until that's deployed, deliver account_link_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:

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:

{
  "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
• platform_health_checks_total
• platform_health_check_errors_total

Local development

Run your Flask app locally:

python app.py

For local testing, expose your server so Pipbit can reach POST /pipbit/agent.

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:

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:

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

The SDK exposes:

• AuthenticationError
• InvalidRequestError
• ConfigurationError
• 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.

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