Installation

pip install pipbit

The SDK requires Python 3.9 or later. It installs Flask as a dependency and provides a lightweight framework for receiving webhook calls from the Pipbit platform and responding with spoken text.

Quick start

from pipbit import Agent

agent = Agent(
    name="my-agent",
    secret="pb_live_sk_your_secret_here",
    base_url="https://api.pipbit.ai",
)

@agent.handle
def handle(req):
    if req.is_connect_intro():
        return "Hello! I'm your agent. How can I help?"
    return f"You said: {req.text}"

agent.run(host="0.0.0.0", port=3002)

The Request object

Your handler receives a Request with these fields:

Field	Type	Description
text	str	The user's utterance, or the connect-intro sentinel on first turn
conversation	list	Recent conversation history (role + text pairs)
subject_id	str	Persistent user identity across sessions
agent_session_id	str	Current agent session identifier
device_shadow	dict	Device state snapshot (sensors, network, location)
device_id	str	Device identifier (always present for environment events)
handoff_context	str	Free-text context from Pip describing why the user was handed off
environment_event	dict	Structured event payload (type, data, timestamp) — only for environment events
reply_token	str	Token for deferred replies via push_reply()

Reply modes

Immediate reply

Return a string from your handler. Pip speaks it immediately.

@agent.handle
def handle(req):
    return "Got it, checking now."

Deferred reply

For longer operations, return req.defer() to acknowledge the turn, then call push_reply() when you have the answer.

@agent.handle
def handle(req):
    threading.Thread(target=do_work, args=(req.reply_token, req.text)).start()
    return req.defer()

def do_work(reply_token, text):
    result = slow_api_call(text)
    agent.push_reply(reply_token, result)

Proactive messaging

Send messages to a user outside of an active conversation. Pip will deliver them when the user is next available.

agent.push_message(
    text="Your garage door has been open for 30 minutes.",
    subject_id="user_abc123",
)

Environment events

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

{"event_subscriptions": ["audio_scene"]}

Events arrive as normal webhook calls. Detect them with req.is_environment_event():

@agent.handle
def handle(req):
    if req.is_environment_event():
        event = req.environment_event
        if req.agent_session_id:
            return "I heard something."       # active session
        else:
            agent.push_message(               # background
                text="Alert from my agent.",
                device_id=req.device_id,
            )
            return req.defer()
    # ... normal handler

When agent_session_id is present your agent has an active conversation — reply directly. When it's None the user is talking to Pip; use push_message() to signal Pip instead.

Health checks

Register health checks so the platform knows if your agent is ready. Required checks must pass for the agent to receive traffic.

def check_api_key():
    if os.getenv("API_KEY"):
        return True, "ok"
    return False, "API_KEY not configured"

agent.add_health_check("api_key", check_api_key, required=True)

Metrics

The SDK automatically tracks request counts. Access them at /pipbit/metrics on your agent's server.