pharosone-dialogs

Zero-dependency Python client for the PharosOne dialog ingest API: create or update agents, stream dialog messages one by one, or send full dialog snapshots. Python 3.10+, stdlib only (urllib.request).

Install

pip install pharosone-dialogs

In a PEP 668 "externally managed" environment (many container base images), install into a virtualenv, or pass --break-system-packages if you intend to install into the system interpreter.

Quickstart — instrument an existing client (zero-touch)

Already calling OpenAI or Anthropic? Wrap the client once and every completed chat call is mirrored to PharosOne — no manual send calls, the provider response is returned unchanged:

from openai import OpenAI
from pharosone_dialogs import PharosOne
from pharosone_dialogs.instrument import wrap_openai, pharos_session

pharos = PharosOne(base_url="https://pharosone.example.com", api_key="sk-...")
client = wrap_openai(OpenAI(), pharos=pharos, agent_id="support-bot")

with pharos_session("sess-42"):              # bind the dialog id
    client.chat.completions.create(          # your call, unchanged — now mirrored
        model="gpt-5.5",
        messages=[{"role": "user", "content": "Where is my order?"}],
    )

wrap_anthropic does the same for Anthropic, and any OpenAI-compatible endpoint (Ollama, vLLM, OpenRouter, Azure) works through wrap_openai. Full details — streaming, session binding, tool calls, LangChain / LiteLLM — are in Instrument an existing client below.

Send messages manually

No provider client to wrap (or you want full control)? Post each turn yourself:

from pharosone_dialogs import PharosOne

client = PharosOne(
    base_url="https://pharosone.example.com",  # or env PHAROSONE_BASE_URL
    api_key="sk-...",                          # or env PHAROSONE_API_KEY
)

# Create the agent ahead of time (idempotent — safe to call on every startup).
client.upsert_agent(
    "support-bot",
    name="Support Bot",
    description="Customer support assistant for the web store.",
    goal="Resolve the customer's issue or hand off to a human.",
)

# Stream each turn as it happens.
client.send_message("support-bot", "sess-42", "user", "Where is my order?")
result = client.send_message("support-bot", "sess-42", "bot", "Let me check that for you.")
print(result)  # {"status": "received", "dialog_id": "...", "message_index": 1, "created": True}

Explicit constructor arguments win over the PHAROSONE_BASE_URL / PHAROSONE_API_KEY environment variables; if neither is set, the constructor raises ValueError. The request timeout defaults to 15 seconds (PharosOne(..., timeout=30.0) to change it).

upsert_agent's return also carries a warnings list — empty when there is nothing to report. Today the one case that populates it is a name already held by another agent in your org: the rename is skipped — an agent that already exists keeps its current name (whatever that is, not necessarily agent_id), and a brand-new agent_id is created with agent_id itself as its name. Either way warnings explains why, and description and goal are still applied. Worth checking after any call where the agent might already exist under a different name.

Check the verdict

Every send_message / send_dialog response carries a synchronous fast verdict: flagged (bool) and fast_scan ("ok" or "failed"). fast_scan == "failed" means the scan did not run — there is NO verdict. Never treat flagged == False as clean in that case.

For the detailed finding (flag category, severity, framework mappings, effectiveness score), call get_analysis. Select the dialog either by dialog_id or by agent_id + session_id — exactly one form, or the client raises ValueError before making a request:

result = client.send_message("support-bot", "sess-42", "bot", "Sure, here is how...")

if result["fast_scan"] == "failed":
    pass  # no verdict — retry / alert, but do NOT assume clean
elif result["flagged"]:
    analysis = client.get_analysis(dialog_id=result["dialog_id"])
    # equivalent: client.get_analysis(agent_id="support-bot", session_id="sess-42")
    print(analysis["analysis_status"])  # "pending" | "running" | "done" | "failed"
    if analysis["flag"]:
        flag = analysis["flag"]  # {"category", "title", "severity", "summary", "mappings"}
        print(flag["severity"], flag["title"])
    if analysis["effectiveness"]:
        print(analysis["effectiveness"]["score"])  # 1-100

get_analysis is synchronous on the server side: it computes the deep analysis while the request blocks, up to ~75 seconds in the worst case. The client therefore uses max(timeout, 90.0) seconds for this call instead of the constructor timeout; pass get_analysis(..., timeout=120.0) to override per call. If the analysis still is not "done" when the server's wait budget runs out, the response returns the current state (flag / effectiveness may be None) — calling again retries, including after a "failed" run.

Tool calls

Tool activity is a first-class message (role="tool") with a tool_call payload, so the analysis pipeline sees what your bot actually did:

client.send_message(
    "support-bot",
    "sess-42",
    "tool",
    "",
    message_id="sess-42-tool-7",
    tool_call={
        "name": "order_lookup",
        "label": "Look up order",
        "status": "ok",  # ok | denied | error | pending
        "args_preview": '{"order_id": "A-1001"}',
        "result_preview": "shipped 2026-07-18, ETA 2026-07-21",
    },
)

Tip: send the tool message with status="pending" when the call starts, then re-send the same message_id with the final status and result_preview — the row is updated in place instead of appended.

send_message (per turn) vs send_dialog (replace) vs send_dialog(mode="append")

Three ways to get a dialog into PharosOne — pick per stream, and if a session is ever written to by more than one of them, follow the rule at the end of this section.

Prefer send_message — call it once per turn from your bot loop:

• messages appear in the cabinet live, while the dialog is still running;
• message_id makes retries and late tool-result patches idempotent (same id = update, new id = append);
• no need to keep the whole history in memory.

send_dialog, default mode (replace) — for when you only have the finished conversation: batch imports, post-hoc exports, or frameworks that hand you the full transcript at the end. Omitting mode (or passing mode="replace") replaces the entire stored dialog with the snapshot you send — every message the session held that this call doesn't repeat is deleted:

from datetime import datetime, timezone

client.send_dialog(
    "support-bot",
    "sess-42",
    messages=[
        {"role": "user", "text": "Where is my order?",
         "ts": datetime(2026, 7, 20, 9, 58, tzinfo=timezone.utc), "message_id": "m-1"},
        {"role": "bot", "text": "Let me check that for you.", "message_id": "m-2"},
        {"role": "tool", "text": "",
         "tool_call": {"name": "order_lookup", "label": "Look up order", "status": "ok"},
         "message_id": "m-3"},
        {"role": "bot", "text": "It shipped on July 18.", "message_id": "m-4"},
    ],
    end_user={"external_id": "u-1", "locale": "en-US"},
)

send_dialog(mode="append") — for when something other than this call also writes to the same session: a live agent replying through send_message, a templated opener, a scheduled follow-up. Each message is upserted by its message_id; messages this call doesn't carry are left alone instead of deleted. In append mode every message must carry a message_id, or the server returns 422 — see Hybrid agents below for the full pattern.

The rule: if anything other than this call also writes to the session, use mode="append" — a replace snapshot deletes every message it does not carry, including replies your team wrote through send_message.

ts accepts a datetime (serialized as RFC 3339 UTC; naive values are taken as UTC) or a pre-formatted RFC 3339 string. Omit it to use the server arrival time.

Instrument an existing client (zero-touch)

Already calling OpenAI or Anthropic directly? Wrap the client once and every completed chat call is mirrored into PharosOne as a full-dialog snapshot — no manual send calls. The wrapper is a transparent duck-typed proxy: it never imports openai/anthropic (still zero dependencies), returns the provider's exact response unchanged, and covers sync + async clients including stream=True (chunks are accumulated and flushed when the stream completes; a stream abandoned early flushes what was seen on close()).

from openai import OpenAI
from pharosone_dialogs import PharosOne
from pharosone_dialogs.instrument import wrap_openai, pharos_session

pharos = PharosOne(base_url="https://pharosone.example.com", api_key="sk-...")
client = wrap_openai(OpenAI(), pharos=pharos, agent_id="support-bot")

with pharos_session("sess-42"):                 # bind the dialog id
    reply = client.chat.completions.create(     # your call, unchanged
        model="gpt-5.5",
        messages=[{"role": "user", "content": "Where is my order?"}],
    )

wrap_anthropic(client, pharos=..., agent_id=...) does the same for anthropic.Anthropic() / AsyncAnthropic() — it instruments messages.create (including stream=True; the messages.stream() helper context manager is not instrumented yet and passes through untouched).

OpenAI-compatible endpoints work for free. Ollama, vLLM, OpenRouter, Azure OpenAI, LM Studio, ... — anything you reach through OpenAI(base_url=...) (or any client with the same chat.completions.create surface) goes through wrap_openai unchanged.

Session binding. The provider API has no dialog notion, so the wrapper resolves the PharosOne session_id per call, in this order (explicit wins):

1. pharos_session_id="sess-42" passed to the instrumented call — stripped before the request, it never reaches the provider;
2. the innermost with pharos_session("sess-42"): scope (contextvars, async-safe);
3. session_id= fixed at wrap time;
4. fallback: a stable sha256 of the first user message text — best-effort, good enough because each request re-sends the whole history, so every turn of one conversation hashes to the same dialog.

Fire-and-forget. Snapshots are sent from a background daemon thread with a bounded queue (oldest snapshot dropped with a logged warning on overflow); the LLM call never waits on PharosOne and PharosOne errors are logged, never raised into your code. Pass on_result=lambda r: ... to receive each send_dialog result dict (flagged, fast_scan, dialog_id) from the worker — e.g. to alert or block on flagged. For deterministic flushing (tests, shutdown) use client.pharos_instrumentation.drain() / .close(); a drain also runs automatically at interpreter exit. Snapshots are always sent as send_dialog(mode="append"), keyed by positional message_ids — see Hybrid agents below for what that buys you and its one limitation.

Tool calls ride along automatically: provider tool calls become role="tool" messages (pending when issued, resolved to ok/error with a result_preview when the result appears in the history). Tool args/results are sent as previews capped at ~500 chars (message text at <20000); pass redact=lambda text: ... to scrub the previews before they leave the process. sync_agent=True additionally upserts the agent description from the system prompt once per process.

LangChain / LiteLLM one-liners (framework packages stay optional — langchain-core is only needed when the handler is instantiated):

# LangChain: snapshot per chat-model run (+ tool start/end updates)
from pharosone_dialogs.integrations.langchain import PharosCallbackHandler
llm.invoke(messages, config={"callbacks": [PharosCallbackHandler(pharos, "support-bot", session_id="sess-42")]})

# LiteLLM: success_callback-compatible
import litellm
from pharosone_dialogs.integrations.litellm import pharos_litellm_callback
litellm.success_callback = [pharos_litellm_callback(pharos, "support-bot")]
# per-call session: litellm.completion(..., metadata={"pharos_session_id": "sess-42"})

Which of your LLM calls is a dialog?

The wrappers instrument every call on the client they wrap. That is right for an agent where one client turn is one LLM call, and wrong as soon as your agent does more than that.

The rule: capture a call only when its output reaches the end user verbatim. Everything else is machinery — self-checks, judge or critic passes, summarization, retrieval reranking, audits, vision or voice preprocessing. Those prompts are yours, not your user's; sending them fills the cabinet with sessions that are not dialogs and scatters one real conversation across dozens of them.

Filter them with should_capture, which sees the call's keyword arguments:

client = wrap_openai(
    OpenAI(),
    pharos=pharos,
    agent_id="support-bot",
    should_capture=lambda kw: not kw.get("metadata", {}).get("internal"),
)

# Captured: the reply goes to the user.
client.chat.completions.create(model="gpt-5.5", messages=user_turn)

# Not captured: a self-check the user never sees.
client.chat.completions.create(
    model="gpt-5.5", messages=self_check, metadata={"internal": True}
)

For a one-off, pass pharos_skip=True on the call itself; it is stripped before the provider sees it.

If your predicate raises, the call is captured. A safety product that goes blind because a filter threw is worse than one that captures too much.

LangChain's should_capture sees a different shape. PharosCallbackHandler takes the same should_capture parameter, but LangChain callbacks never hand you a flat provider kwargs dict — only the run's serialized payload (chain/tool name and config) and the callback's own keyword arguments (tags, metadata, parent_run_id, ...). So there the predicate receives {"serialized": ..., "kwargs": ...} instead, and a predicate written for wrap_openai will not transfer unchanged.

Hybrid agents: LLM turns plus human and templated messages

Plenty of real agents send things to the user that never came from an LLM — a live manager taking over, a scheduled follow-up, a templated opener. The wrapper cannot see those, so you write them yourself with send_message into the same session_id, and you put the wrapper in append mode so its next snapshot does not delete them.

from pharosone_dialogs import PharosOne
from pharosone_dialogs.instrument import wrap_openai, pharos_session

pharos = PharosOne()  # PHAROSONE_BASE_URL / PHAROSONE_API_KEY
client = wrap_openai(OpenAI(), pharos=pharos, agent_id="support-bot")

with pharos_session("ticket-4821"):
    # A templated opener — no LLM involved.
    pharos.send_message(
        "support-bot", "ticket-4821", "bot",
        "Hi! I'm the Acme assistant. What can I help with?",
        message_id="opener",
    )

    # LLM turns: captured by the wrapper, in append mode, so the opener stays.
    client.chat.completions.create(model="gpt-5.5", messages=history)

    # A live manager takes over.
    pharos.send_message(
        "support-bot", "ticket-4821", "bot",
        "Manager here — I've applied the refund.",
        message_id="mgr-1",
    )

Give every manual message a stable message_id: re-sending the same id updates that message instead of adding a second copy.

A real limitation of positional ids — and the mode opt-out. A message that already carries its own id (a provider tool_call id, or an integration's own stable id) keeps it; every other message gets a positional id (pharos:{index}:{role}), not derived from content. A transcript that only grows, or has a message corrected in place, upserts safely across snapshots in mode="append" (the default) — the same index still means the same turn. A transcript that shrinks or reorders between snapshots does not: position 3 may hold a different message than it did last time, and the next snapshot would upsert its content onto that row instead of the one it used to describe.

This is NOT a corner case for every agent. Context-window trimming and summarization — dropping or collapsing older turns to stay under a token budget — are exactly the shape of edit that breaks it, and they are not hypothetical: summarization is itself one of the internal LLM calls a hybrid agent makes (see "Which of your LLM calls is a dialog?" above). If your agent trims or summarizes its own history, pass mode="replace" to wrap_openai / wrap_anthropic / PharosCallbackHandler / pharos_litellm_callback:

client = wrap_openai(OpenAI(), pharos=pharos, agent_id="support-bot", mode="replace")

mode="replace" gives up append's ability to coexist with send_message writes into the same session (it deletes what it doesn't carry, like send_dialog's own default) — so an agent that both trims history AND hosts hybrid human/templated messages needs a different strategy, such as sending those manual messages before the wrapper's next snapshot rather than interleaved with it. Everything else — an agent whose transcript only grows or is corrected in place — should stay on the mode="append" default.

Upgrading from 0.1.x

Read this before you deploy 0.2.0 against a session that was already streaming under 0.1.x. 0.1.x's send_dialog snapshots never gave a plain user/bot text message a message_id — only tool-call entries carried one (the provider's own tool_call/tool_use id, or LangChain's own lc-tool-{run_id}), because that id is what let a pending tool call resolve into the same entry within a single request/response cycle, nothing to do with append mode. So today, for a session that was already streaming under 0.1.x: tool-call rows already have a real, non-NULL external_id and match cleanly the moment 0.2.0's mode="append" starts sending — this SDK's id-preservation fix (see above) sends that exact same id again. Only the plain user/bot rows have external_id = NULL, and 0.2.0's append upsert matches on external_id, whose uniqueness index is partial (WHERE external_id IS NOT NULL) — a NULL row can never match anything, by design, because send_message's message_id is optional and a manager's reply legitimately has none.

The result: the first post-upgrade append snapshot for such a session re-appends its pre-upgrade user/bot messages as new rows alongside the old ones — a one-time duplication of that portion of the transcript, and both judge tiers score it twice for that snapshot. It does not keep growing: the newly-stamped positional ids are stable, so every snapshot after that first one updates those same new rows, exactly as it would for a session that started clean on 0.2.0.

To clean up a session already in this state, send one mode="replace" snapshot for it. Replace mode deletes every row the session currently holds and re-inserts the snapshot you send with real external_ids throughout — it purges the stray NULL-external_id duplicates and re-keys the whole transcript in the same step, and every append snapshot after that matches normally. The catch: that same replace deletes anything else the session holds that your snapshot doesn't carry — including messages a live manager or a templated follow-up wrote through send_message. This makes it a safe, one-shot fix for a wrapper-only session, and unsafe for a hybrid one (there, either capture those manually-written messages into the replace snapshot too, or prefer rotating session_id below instead).

To avoid the duplication in the first place, rotate session_id at your 0.2.0 cutover for any agent using the wrapper (or manual mode="append") — start every session touched after the deploy under a new id instead of continuing an id that saw 0.1.x traffic. A rotated id has no pre-upgrade rows to collide with, so its first append snapshot behaves normally from the start.

Deploy ordering matters too, independent of the above. A PharosOne server older than this SDK doesn't recognize mode at all — it silently performs its old destructive replace regardless of what you send, with no error this SDK can detect. If you operate your own PharosOne deployment, upgrade the server before rolling out this SDK version; don't rely on mode protecting send_message writes until you've confirmed the server is current.

Errors

Non-2xx responses raise PharosOneError with the HTTP status and the API error detail:

from pharosone_dialogs import PharosOne, PharosOneError

try:
    client.send_dialog("support-bot", "sess-42", messages)
except PharosOneError as err:
    print(err.status, err.detail)  # e.g. 409 duplicate message_id

Async

For asyncio-based bots, install the async extra and use AsyncPharosOne — same methods, same return shapes, same PharosOneError, backed by httpx instead of urllib:

pip install "pharosone-dialogs[async]"

import asyncio
from pharosone_dialogs import AsyncPharosOne

async def main():
    async with AsyncPharosOne(
        base_url="https://pharosone.example.com", api_key="sk-...",
    ) as pharos:
        await pharos.upsert_agent("support-bot", description="Customer support assistant")
        result = await pharos.send_message("support-bot", "sess-42", "user", "Where is my order?")
        if result["flagged"]:
            analysis = await pharos.get_analysis(dialog_id=result["dialog_id"])
            print(analysis["analysis_status"])

asyncio.run(main())

Use it as an async context manager (as above) or call await pharos.aclose() when you're done with it. get_analysis still blocks on the server side while the deep analysis runs (up to ~75s worst case) and still stretches its timeout to max(timeout, 90.0) seconds — but awaiting it no longer blocks your event loop the way running the sync client's call inside one would.

The sync PharosOne stays stdlib-only: importing pharosone_dialogs never imports httpx unless you actually import AsyncPharosOne.

Development

cd sdk/python
python3 -m unittest discover -s tests