WhatsApp transport and LLM flow logic, extracted once.

Channel & InboundMessage

Defined in joiinflow.whatsapp.types, re-exported from joiinflow.whatsapp.

Channel

class Channel(str, Enum):
    META = "meta"
    TWILIO = "twilio"

Which transport a message arrived on / a reply should go out on. Replaces the old agency.get("_twilio") dict-sniffing hack with something typed and switchable.

InboundMessage

A normalized inbound WhatsApp message, regardless of transport — the output of webhook parsing and the input to FlowHandler.handle_inbound().

Design principles

These aren't style preferences — each one closes a specific problem the two duplicated handlers had.

Explicit config, no globals

Every class takes JoiinflowConfig (and, for storage, a store/client instance) in its constructor. Nothing reaches for an environment variable or a cached singleton behind your back. You can run two differently-configured FlowHandlers in the same process if you need to.

Protocols, not implementations, for persistence

SessionStore is a typing.Protocol. joiinflow ships two implementations (InMemorySessionStore, RedisSessionStore), but it never touches Supabase, Postgres, or any app-specific table — that stays behind FlowHandler's on_inbound_received/on_reply_sent hooks, which you implement.

The four-hook subclassing boundary

FlowHandler owns the send/receive/history/tool-loop/error-handling sequence. You implement build_system_prompt, build_tools, on_tool_call, and optionally on_inbound_received/on_reply_sent — that's the entire surface where domain logic plugs in. See FlowHandler.

Typed for autocomplete

Ships py.typed (PEP 561) with full annotations. Once installed, Pylance/mypy give real completions on every joiinflow.* import — no stub package needed.

TwilioChannel

joiinflow.whatsapp.twilio.TwilioChannel — Twilio BYON (Bring Your Own Number) WhatsApp: sending, subaccount management, and OTP-based number registration. This is the transport half only; DB-touching persistence (which subaccount belongs to which agency, usage-log rows) is deliberately left to your app.

from joiinflow.whatsapp import TwilioChannel
channel = TwilioChannel(master_account_sid, master_auth_token)  # needed for subaccount mgmt only

Signature verification

joiinflow.whatsapp.signatures. Verifying inbound webhook POSTs is not optional — anyone who discovers your webhook URL can otherwise post forged messages that get treated as real customer conversations.

async def receive_webhook(request: Request) -> dict:
    raw_body = await request.body()  # read raw bytes BEFORE .json()
    signature = request.headers.get("X-Hub-Signature-256", "")
    if not verify_meta_signature(settings.meta_app_secret, signature, raw_body):
        return {"received": False}
    payload = await request.json()
    ...

Webhook parsing

joiinflow.whatsapp.webhook. Normalizes each transport's webhook payload shape into InboundMessage — route registration itself stays app-side.

Session storage

joiinflow.flow.session. Replaces the module-level _history_store: dict global both apps used — a plain in-process dict silently loses history on every Vercel cold start, since there's no guarantee two requests for the same conversation hit the same warm process.

SessionStore protocol

class SessionStore(Protocol):
    def get_history(self, key: str) -> list[dict]: ...
    def save_history(self, key: str, history: list[dict]) -> None: ...
    def clear_history(self, key: str) -> None: ...

Structural — anything with these three methods works, no inheritance required.

InMemorySessionStore(max_history=20)

A plain dict under the hood. Fine for local dev and tests. Not recommended in production on Vercel — same cold-start problem as before.

RedisSessionStore(redis_client, ttl_seconds=86400, max_history=20, key_prefix="conv")

The recommended default. Survives cold starts because history lives in Redis, not process memory. redis_client must satisfy RedisLike:

class RedisLike(Protocol):
    def get(self, key: str): ...
    def set(self, key: str, value: str, ex: int | None = None): ...
    def delete(self, key: str): ...

Both upstash_redis.Redis and redis-py's redis.Redis satisfy this out of the box — construct whichever your app already uses and pass the instance in. joiinflow never builds the connection itself.

from upstash_redis import Redis
from joiinflow.flow import RedisSessionStore

store = RedisSessionStore(Redis(url=settings.upstash_redis_url, token=settings.upstash_redis_token),
                           ttl_seconds=settings.redis_ttl_seconds)

Keys are stored as {key_prefix}:{key}; ttl_seconds is passed as Redis's ex on every save, so a conversation's history simply expires if nobody replies for that long — no separate staleness bookkeeping needed.

ToolSpec & ToolRegistry

joiinflow.flow.tools. Generalizes the ad-hoc _P24_TOOL/_CALENDAR_TOOLS dict-literal pattern from the original handlers into a typed shape.

ToolSpec

ToolSpec(name: str, description: str, input_schema: dict)

input_schema is a standard JSON Schema object, exactly as Anthropic's tool-use API expects. .to_anthropic_schema() returns {"name", "description", "input_schema"} — Anthropic's wire format.

SEARCH_TOOL = ToolSpec(
    name="search_property24",
    description="Search the agency's Property24 listings...",
    input_schema={
        "type": "object",
        "properties": {
            "area": {"type": "string"},
            "budget": {"type": "string"},
        },
        "required": ["area", "budget"],
    },
)

ToolRegistry(tools=None)

An ordered collection you rarely construct directly — FlowHandler.handle_inbound wraps whatever list your build_tools hook returns in one internally. bool(registry) is False for an empty registry (that's the "empty list ⇒ plain chat" check), .to_anthropic_tools() returns the list achat_with_tools needs, .names() lists registered tool names.

Deploy to Vercel

Deployment is unchanged by using joiinflow — it's a library swap, not a deploy-shape change. The one new piece is that Vercel's build now needs to clone a private dependency.

Checklist

1. Connect the repo to a Vercel project as usual (framework auto-detects as fastapi from vercel.json).
2. Set every env var your Settings class reads (Meta/Twilio/Anthropic/Supabase/Redis credentials) in the Vercel project's Environment Variables.
3. Give the Vercel build a credential that can clone kagisano/joiinflow-sdk — a fine-grained GitHub PAT (read-only, scoped to that one repo) is the simplest option. Vercel doesn't have a built-in "private Python dependency" credential slot the way it does for private npm registries, so this typically means baking the PAT into the git URL via a build-time env var, e.g. rewriting the dependency to git+https://${GITHUB_PAT}@github.com/kagisano/joiinflow-sdk.git@<ref> at build time, or configuring a global git credential helper in a custom build command.
4. Deploy a preview build and check the build logs for the uv sync / pip install step specifically — confirm it actually resolves joiinflow rather than silently failing and falling back to a cached wheel from a previous successful build.

uv.lock

Both apps commit uv.lock so Vercel's build resolves exactly the dependency graph you tested locally, rather than re-resolving (and potentially picking up a newer, untested joiinflow commit if you're pinned loosely, or a newer transitive dependency).

Test locally & in VS Code

Set up the interpreter

uv sync
cp .env.example .env   # fill in real values

In VS Code: Cmd/Ctrl+Shift+P → Python: Select Interpreter → pick ./.venv/bin/python. Because joiinflow ships py.typed, Pylance gives full autocomplete and type-checking on every joiinflow.* import immediately — no separate stub package needed.

Run the server

uv run uvicorn app.main:app --reload --port 8000

Open http://localhost:8000/docs — FastAPI's Swagger UI lists every route and lets you fire test requests without needing WhatsApp at all. Endpoints that don't touch your database (e.g. a GET /webhook handshake, a GET /debug/... route) work with zero external credentials configured.

Testing real inbound messages before deploying

ngrok http 8000

Register the resulting https://....ngrok-free.app/webhook URL as your Meta app's webhook callback URL, message your WhatsApp test number from a phone, and watch it flow through your local server with breakpoints available.

Testing flow logic without hitting WhatsApp at all

Build a small test-only endpoint (or script) that calls flow_handler.generate_reply(context, history) directly — it runs your prompt/tool logic and returns the text, with no send and no session-store write. This is exactly what joiinflow-backend's POST /test-chat does for its dashboard's "Test Flow" panel.

Iterating on the SDK itself alongside an app

Instead of pushing to joiinflow-sdk and re-pinning a commit SHA on every change while you're iterating:

uv add --editable ../joiinflow-sdk

This points the app's dependency at your local checkout — edits show up immediately, no reinstall. Revert to the git URL before committing so CI and Vercel still resolve the real, pushed package.

Debug config (.vscode/launch.json)

{
  "version": "0.2.0",
  "configurations": [{
    "name": "FastAPI",
    "type": "debugpy",
    "request": "launch",
    "module": "uvicorn",
    "args": ["app.main:app", "--reload", "--port", "8000"],
    "envFile": "${workspaceFolder}/.env",
    "console": "integratedTerminal"
  }]
}

Migrate an existing app

Two real apps have already gone through this. Both are worth reading as case studies before migrating a third.

Migration checklist

1. Add the git dependency to pyproject.toml, run uv sync.
2. Turn your existing transport files (whatsapp.py, anthropic_service.py) into thin adapters: construct the joiinflow class from your app's config, re-export the same function names your call sites already use. This keeps every existing caller working unchanged — you're swapping the implementation underneath, not the call sites.
3. Split your existing flow handler: identify what's generic orchestration (delete it, FlowHandler replaces it) vs. domain-specific (loaders, prompt builder, tool dispatch — move into your FlowHandler subclass's methods, logic unchanged).
4. Swap your session store from whatever in-process dict it was to RedisSessionStore.
5. Add signature verification to your webhook routes if it wasn't already there.
6. Keep vercel.json and api/index.py untouched — this is a library swap, not a deploy change.
7. Smoke-test: does the app still import? Does /docs load? Does a scripted "mock the DB, real Claude call" test (see joiinai-li/backend/test_flow_handler.py for the pattern) still produce the same prompt/reply shape as before the migration?

Troubleshooting & FAQ

“uv sync fails to fetch joiinflow”

Check that git clone https://github.com/kagisano/joiinflow-sdk.git works in a plain terminal first, outside of uv entirely. If that fails too, it's a Git credential problem, not a uv problem — see Authenticating to a private repo. If it succeeds but uv sync still fails, run with -v for the underlying git error.

“Signature verification always fails”

By far the most common cause: you called await request.json() before computing the signature, or otherwise hashed something other than the exact raw bytes Meta/Twilio sent. Call request.body() first, verify, then parse. See the callout in Signature verification.

“RedisSessionStore isn't persisting between messages”

Confirm session_key is actually stable across turns from the same customer — a common bug is building it from something that changes per-request (a message ID, a timestamp) instead of a stable pair like agency_id:phone_number.

“Should I use chat_with_tools or achat_with_tools?”

If you're driving a turn through FlowHandler, you never call either directly — handle_inbound calls achat_with_tools for you via your on_tool_call hook. Call AnthropicChatClient methods directly only for standalone use outside the flow engine (e.g. a one-off classification call) — use chat_with_tools there if you're in synchronous code, achat_with_tools if you're already in an async function.

“How do I test the flow logic without real WhatsApp/Twilio/Supabase credentials?”

Mock the DB-touching functions in your handler module (unittest.mock.patch) and call flow_handler.generate_reply(context, history) or drive handle_inbound with a fake SessionStore/channel object satisfying the relevant protocol. See joiinai-li/backend/test_flow_handler.py for a working example that patches DB loaders with fixture data and only needs ANTHROPIC_API_KEY to exercise a real Claude call.

“on_inbound_received returned True but I still see a reply sent”

If your hook sends its own reply (e.g. a scripted state-machine message) before returning True, make sure you're calling self._send(context, to, text) — not accidentally also relying on handle_inbound's normal send path, which only runs when the hook returns False.

“I need a package version tag, not a commit SHA”

See the callout in Install & versioning — no tag has been cut yet. Ask whoever has push access to kagisano/joiinflow-sdk to run git tag v0.1.0 <sha> && git push origin v0.1.0, which will also trigger the release workflow (lint/type-check/test → build sdist+wheel → GitHub Release).

“Where do I add a new WhatsApp transport (e.g. a third provider)?”

Add a new module under joiinflow/whatsapp/ alongside meta.py/twilio.py, add a matching value to the Channel enum, and extend FlowHandler._send's dispatch (currently an if channel == Channel.TWILIO: ... elif channel == Channel.META: ... chain) to route to it. This is a change to the SDK itself, not something app code should work around.

Versioning & releases

SemVer git tags (vX.Y.Z) on joiinflow-sdk. The package version is derived automatically from the tag via hatch-vcs — nobody hand-edits a version number in pyproject.toml.

Release workflow

Pushing a tag matching v*.*.* triggers GitHub Actions: lint (ruff) → type-check (mypy) → test (pytest) → build sdist + wheel → publish a GitHub Release with those artifacts attached. A separate CI workflow runs the same lint/type-check/test steps on every push to main and every pull request, independent of tagging.

Consuming a new version

Bump the @<ref> in each consuming app's pyproject.toml dependency line to the new tag, run uv sync to update uv.lock, and commit both files together in one PR. Since FlowHandler subclasses only touch the four public hooks and a small, deliberately-stable constructor signature, most SDK releases should be non-breaking for existing subclasses — treat any change to hook signatures or the context dict contract as a breaking (major) change.