Integrating the SDK

Connect any channel — a web UI, Zendesk, Slack, an IVR — to a running agent with the InteractiveAI Python SDK: sessions, messages, both event delivery modes, handover, and production patterns.

Context — This guide connects your channel to a running agent using the interactiveai Python SDK, in the order you'll write the code. (Not sure which integration surfaces you need? Start with the Integration overview.) It covers the agent-integration surface; the full SDK API reference (every class, method, and parameter) lives in the InteractiveAI Python SDK documentation — this page links the concepts, the SDK docs hold the signatures. You need the agent's base URL and its API key (the manifest's runtime.api_key value).

1. Install

pip install "interactiveai[agent]"

The [agent] extra pulls in the async client used to talk to the agent server.

2. Construct the client

import asyncio

from interactiveai.agent import InteractiveAgentClient


async def main() -> None:
    async with InteractiveAgentClient(
        base_url="https://my-agent.example.com",
        api_key="<AGENT_API_KEY>",
    ) as client:
        sess = await client.sessions.open(id="customer-1")
        await sess.post_user_message("Hello!")


asyncio.run(main())

Use it as an async context manager so the underlying HTTP client closes cleanly. In a long-lived service, construct one client at startup and share it — it holds a connection pool, and there's no per-session cost to keeping it alive.

3. Open a session

Pick one stable id per end-user (a Zendesk user id, a signed cookie value, an internal account id) and hand it to sessions.open. The SDK registers the customer if needed, finds their session, or creates a new one — the call is idempotent, so call it freely:

sess = await client.sessions.open(
    id="zendesk-42",
    variables={"plan": "pro", "region": "EU"},
)

Variables vs metadata — two different things, easy to mix up:

• Variables — agent-visible context, surfaced on the next turn. This is how you tell the agent who the customer is or what just happened. Any JSON-serialisable value works.

• Metadata — opaque key/value storage on the customer or session. The agent does not see it. Use it for your own bookkeeping: channel info, external ids, anything you'll look up later.

Full model: Sessions, memory & state.

Need more than one session per customer (a "new conversation" button, separate threads per topic)? Use the customer path — client.customers.register(id=..., name=...) then customer.new_session(). Both paths converge on the same customer record; mix them freely.

4. Post a user message

await sess.post_user_message("Hi, I can't log in.")

That single call wakes the engine. The reply comes back as events, not as a return value.

5. Receive replies — pick a delivery mode

Mode	Use when
Polling (default)	Your receiver can hold a long-lived connection — typically a browser-backed UI.
Webhook	Server-to-server channels (Zendesk, Slack, IVR). The agent POSTs events to a URL you expose.

Polling — stream events with sess.events()

from interactiveai.agent import (
    AssistantMessage,
    Preamble,
    StatusEvent,
    ToolEvent,
)

async for ev in sess.events():
    if isinstance(ev, AssistantMessage):
        print("agent:", ev.text)
    elif isinstance(ev, Preamble):
        # Filler the agent emits while still working — render like a
        # typing indicator with text, not as a final reply.
        print("agent (preamble):", ev.text)
    elif isinstance(ev, ToolEvent):
        for call in ev.tool_calls:
            print("tool:", call.tool_id, call.arguments)
    elif isinstance(ev, StatusEvent) and ev.status == "error":
        print("agent error:", ev.error_detail)

sess.events() opens a long-poll stream and reconnects automatically on transient failures. The loop is open-ended — in a real integration you iterate for the life of the channel and break only on shutdown.

Webhook — let the agent push events to you

Pass webhook= at construction:

async with InteractiveAgentClient(
    base_url="https://my-agent.example.com",
    api_key="<AGENT_API_KEY>",
    webhook="https://my-integration.example.com/agent-events",
) as client:
    sess = await client.sessions.open(id="zendesk-42")
    await sess.post_user_message("Hi, I can't log in.")

The SDK writes that URL into session.metadata["event_webhook_url"] on every session it touches; the agent then POSTs its own output events there — AssistantMessage, Preamble, ToolEvent, StatusEvent — authenticated with the same bearer token. Customer messages are never delivered via webhook; read them back by polling the session's events instead. Your receiver:

from fastapi import FastAPI, Request, Response

from interactiveai.agent import AssistantMessage, parse_webhook_event

app = FastAPI()

AGENT_API_KEY = "devsecret"


@app.post("/agent-events")
async def on_agent_event(request: Request) -> Response:
    if request.headers.get("Authorization") != f"Bearer {AGENT_API_KEY}":
        return Response(status_code=401)

    session_id = request.headers["x-session-id"]   # stamped by the agent
    event = parse_webhook_event(await request.json())
    if event is None:
        return Response(status_code=200)           # unknown kind — drop

    if isinstance(event, AssistantMessage):
        await send_to_zendesk(session_id, event.text)

    return Response(status_code=200)

parse_webhook_event returns the same typed union sess.events() yields, so dispatch code for agent-output events is identical in both modes — just remember webhook mode never delivers UserMessage.

6. Know your events

Event	Meaning
UserMessage	The customer sent a message. Delivered via polling only — never via webhook.
AssistantMessage	The agent's reply.
Preamble	Optional filler emitted while the agent is still working on a turn — a typing indicator with text.
StatusEvent	Engine lifecycle: acknowledged, typing, processing (with stage), ready, cancelled, error.
ToolEvent	One batch of tool calls the engine ran — render "checking inventory…" chips.

Every event has an integer offset. Track the highest you've delivered, then pass min_offset=last + 1 on reconnect to resume exactly where you left off — no replay, no skipped events. Wire-level field reference: Events & callbacks.

Status events drive your UI

Status	Meaning
acknowledged	Engine received the user message.
typing	Show a typing indicator.
processing	Pre-message work; combine with ev.stage for a sub-label.
ready	Turn finished — clear all indicators.
cancelled	The turn was superseded (user sent a new message first).
error	Engine error — read ev.error_detail.

7. Replay history when a UI reconnects

messages = await sess.messages(limit=200)
for m in messages:
    print(m.source, m.content)   # source: "customer" | "ai_agent" | "human_agent" | "human_agent_on_behalf_of_ai_agent"

last_offset = max((m.offset for m in messages), default=-1)

async for ev in sess.events(min_offset=last_offset + 1):
    handle(ev)   # same dispatch as section 5

8. Human handover

When a human takes over in your channel, switch the session to manual mode so the engine stops auto-replying, and record the human's messages in the same session:

await sess.set_manual_mode()
await sess.post_human_agent_message(
    "I've reset your password — try logging in now.",
    agent_name="Sara from Support",
)

# Hand back to the agent later:
await sess.set_automatic_mode()

post_system_message covers everything the customer didn't type — inactivity notices, channel switches. It posts under the human_agent source (there's no separate system source) so it reads back like an operator message and never triggers a bot reply:

await sess.post_system_message("Conversation moved from email to chat.")

9. Handle errors

All failures surface as typed exceptions — you never see raw HTTP-library errors:

Exception	When
AuthError	401/403 — bad bearer token.
NotFoundError	404 — session/customer doesn't exist.
ConflictError	409/422 — duplicate or invalid payload.
ServerError	5xx after retries exhausted.
TransportError	DNS, timeout, connection reset.

from interactiveai.agent import NotFoundError, TransportError

try:
    sess = await client.sessions.get(id=session_id)
except NotFoundError:
    sess = await client.sessions.open(id=external_id)
except TransportError as exc:
    log.warning("agent unreachable: %s", exc)
    raise

10. Minimal end-to-end

import asyncio
import os

from interactiveai.agent import (
    AssistantMessage,
    InteractiveAgentClient,
    StatusEvent,
)


async def main() -> None:
    async with InteractiveAgentClient(
        base_url=os.environ["INTERACTIVE_AGENT_BASE_URL"],
        api_key=os.environ["AGENT_API_KEY"],
    ) as client:
        sess = await client.sessions.open(id="[e-mail]")
        await sess.post_user_message("Can I rent a van for Saturday?")

        async for ev in sess.events():
            if isinstance(ev, AssistantMessage):
                print("agent:", ev.text)
            elif isinstance(ev, StatusEvent) and ev.status == "ready":
                break


asyncio.run(main())

Good for a smoke test — real integrations stay in the event loop indefinitely.

Common patterns

Resume a UI session from a cookie

sessions.open is idempotent — every HTTP handler opens-or-resumes with the same id:

@app.post("/chat")
async def chat(req: Request, session_cookie: str = Cookie(...)) -> Response:
    sess = await client.sessions.open(id=session_cookie)
    body = await req.json()
    await sess.post_user_message(body["message"])
    return Response(status_code=202)

Route an inbound third-party webhook into a session

@app.post("/zendesk/webhook")
async def on_zendesk(req: Request) -> Response:
    payload = await req.json()
    customer_id = f"zendesk-{payload['user']['id']}"
    sess = await client.sessions.open(
        id=customer_id,
        name=payload["user"]["name"],
        variables={"plan": payload["user"]["plan"]},
    )
    await sess.post_user_message(payload["comment"]["body"])
    return Response(status_code=202)

Find an existing session by an external id

Stamp your channel's conversation id into metadata at creation, look it up later instead of guessing:

customer = await client.customers.register(id=user_id, name=user_name)

sess = await customer.find_session(external_id="chatwoot-conv-789")
if sess is None:
    sess = await customer.new_session(
        title="Chatwoot #789",
        metadata={"external_id": "chatwoot-conv-789"},
    )

Push context into the conversation mid-flight

Agent should know something the customer didn't say? Write variables (visible next turn):

await client.customers.set_variables(
    customer.id,
    {"loyalty_tier": "gold", "open_tickets": 3},
)

Bookkeeping only? Write metadata (invisible to the engine):

await sess.update_metadata({"chatwoot_conversation_id": "789"})

Large payloads the agent should treat as fetched data (statements, history dumps) go in as injected tool events instead — see Tools.

Render tool calls in the UI

if isinstance(ev, ToolEvent):
    for call in ev.tool_calls:
        ui.show_chip(f"🔧 {call.tool_id}")
        if call.result is not None:
            log.debug("tool result", tool_id=call.tool_id, result=call.result)

Tune the event stream

sess.events() exposes reconnect/long-poll knobs; defaults suit browser-style UIs:

Parameter	Default	What it does
min_offset	auto	First offset to deliver; None resumes from sess.last_event_offset + 1.
max_reconnect_attempts	5	Transient failures before giving up.
reconnect_initial_delay	1.0s	Initial backoff after a failure.
reconnect_backoff	2.0×	Backoff multiplier.
idle_reconnect_delay	1.0s	Pause before reopening an idle-closed stream.
wait_for_data	60s	Server-side long-poll window.

Production notes

• Same client across replicas. The webhook URL travels on session metadata, not client memory — any replica's engine work finds it.

• Idempotency by offset. Both delivery modes carry offset; persist the largest processed per session and dedupe on (session_id, offset) — at-least-once becomes effectively exactly-once.

• Webhook retries. 429 and 5xx from your /agent-events endpoint (plus connection/timeout errors) trigger a retry with exponential backoff. Other 4xx responses (400/401/403/404) are treated as permanent and are not retried — return 200 once durably enqueued, process in the background.

• Reuse the client; or bring your own HTTP client (custom transport / instrumentation) via the httpx_client= constructor parameter — the SDK adds its auth without clobbering your hooks.

• Don't poll aggressively. The events endpoint is server-side rate-shaped: rapid identical polls get a delayed response instead of an immediate one. Long-poll with sess.events() rather than busy-looping your own polling.

SDK surface map

What you'll touch, and where it's used above — full signatures in the SDK reference documentation:

Entry point	Used for
client.sessions — open, get, get_metadata, update_metadata, set_mode	§3, §8
client.customers — register, retrieve, get_variables, set_variables, get_metadata, update_name	§3, patterns
Customer — new_session, get_session, latest_session, find_session, list_sessions	§3, patterns
Session — post_user_message, post_human_agent_message, post_system_message, set_manual_mode, set_automatic_mode, update_metadata, messages, events, refresh	§4–§8
Event types — UserMessage, AssistantMessage, Preamble, StatusEvent, ToolEvent, ToolCall; parse_webhook_event	§5–§6
Exceptions — InteractiveAgentError, AuthError, NotFoundError, ConflictError, ServerError, TransportError	§9

post_human_handover still exists as a deprecated alias for post_human_agent_message (same signature) — it raises a DeprecationWarning; use post_human_agent_message directly in new code.