flycanon/sdks/python/README.md at main · firefly-operationOS/flycanon

Python SDK

Async-first Python client for the flycanon Operational Knowledge Repository service.

Pydantic-typed request + response models.
httpx under the hood -- pooled connections, retries, timeouts you control.
Carries the firefly four-header wire contract (X-Tenant-Id / X-Workspace-Id / X-Correlation-Id / X-Agent-Token) on every outbound request.
No service dependency -- the SDK ships its own Pydantic schemas so it can be installed alongside any client codebase without pulling the framework.

Wire-contract compatibility

Compatible with flycanon service version 26.7.x.

SDK	Service
`26.8.0`	`26.8.x` (embedding sets, `flycanon reindex`, `ChunkStats.by_embedding_model`)
`26.7.1`	`26.7.x` (routes realigned: `?mode=async`, `/query/stream`, `/conversations/{id}/turn`, `/query/suggest`, `?after_id=`; `X-API-Key`; `delete_source`, `purge_workspace`)
`26.5.7`	`26.5.x`

Install

uv add flycanon-sdk==26.8.0
# or
pip install flycanon-sdk==26.8.0

Quick start

import asyncio
from flycanon_sdk import AnswerRequest, CanonClient


async def main() -> None:
    async with CanonClient(
        base_url="https://canon.example.com",
        tenant_id="acme",
        workspace_id="ws-prod",
        agent_token="canon_xxxxx",
    ) as client:
        result = await client.agent.query(
            AnswerRequest(question="What's our retention policy?"),
            idempotency_key="example-001",
        )
        print(result.answer)


asyncio.run(main())

See QUICKSTART.md for the longer five-minute tour.

Constructor reference

CanonClient(...) -- all kwargs are keyword-only.

Required

kwarg	type	description
`base_url`	`str`	Service base URL (e.g. `https://canon.example.com`).

Optional wire-contract headers

kwarg	type	header sent	notes
`tenant_id`	`str \| None`	`X-Tenant-Id`	Required by the service on every route; pass it here so you don't have to thread it through every call.
`workspace_id`	`str \| None`	`X-Workspace-Id`	Required by the service on every route.
`correlation_id`	`str \| None`	`X-Correlation-Id`	Sticky per-request id. Usually you'll override this per call.
`agent_token`	`str \| None`	`X-Agent-Token`	Only meaningful on `/api/v1/agent/*` routes.

The SDK only injects a header when you pass a value -- it does not pre-validate; the service rejects missing tenant/workspace at the boundary.

Advanced

kwarg	type	description
`api_key`	`str \| None`	Platform key (`FLYCANON_API_KEYS`) sent as `X-API-Key`. Not needed on `/api/v1/agent/*` calls that carry `agent_token`.
`timeout`	`float`	httpx timeout in seconds (default `60`).
`client`	`httpx.AsyncClient \| None`	Caller-supplied client to reuse. When provided, the SDK does not close it on exit.
`headers`	`dict[str, str] \| None`	Extra headers merged into every request.

Method catalog

Workspace CRUD (`/api/v1/workspaces`, user-tier)

Method	Description
`create_workspace(spec: WorkspaceCreate) -> WorkspaceSpec`	Create a workspace under the caller's tenant.
`list_workspaces() -> list[WorkspaceSummary]`	List every workspace owned by the caller's tenant (`created_at DESC`).
`get_workspace(workspace_id) -> WorkspaceSpec`	Fetch a single workspace by id.
`update_workspace(workspace_id, patch: WorkspaceUpdate) -> WorkspaceSpec`	Sparse PATCH; only present fields are applied.
`close_workspace(workspace_id) -> WorkspaceSpec`	Terminal lifecycle transition (`status=closed`, `closed_at=now()`). Idempotent.

Agent token CRUD (`/api/v1/agent-tokens`, user-tier)

Method	Description
`mint_agent_token(request: AgentTokenMintRequest) -> AgentTokenCreated`	Mint a new token. Raw `token` is returned once.
`list_agent_tokens() -> list[AgentTokenSummary]`	List tokens for the caller's tenant (newest first). Secret omitted.
`revoke_agent_token(token_id) -> None`	Revoke a token. Idempotent.

Agent-tier callers (authenticated with X-Agent-Token) are refused on the mint path with 403 agent_cannot_mint.

Agent surface (`/api/v1/agent/*`, agent-tier, exposed via `client.agent`)

Every POST on this surface mandates a non-empty idempotency_key keyword argument. The SDK rejects empty / whitespace values locally with ValueError before the round-trip.

Method	Description
`agent.ingest_source(spec, *, idempotency_key) -> SourceRecord`	Submit a source for intake. Scope: `agent.sources:ingest`.
`agent.get_source(source_id) -> SourceRecord`	Read a source. Scope: `agent.sources:read`.
`agent.query(request, *, idempotency_key) -> AnswerResponse`	Grounded RAG answer with citations. Scope: `agent.query:run`.
`agent.query_stream(request, *, idempotency_key) -> AsyncIterator[IngestJobEvent]`	SSE: `hit` frames + terminal `final`. Scope: `agent.query:run`.
`agent.search(request, *, idempotency_key) -> SearchResponse`	Hybrid retrieval, no LLM. Scope: `agent.query:run`.
`agent.get_knowledge(item_id) -> KnowledgeItem`	Fetch a single knowledge item. Scope: `agent.knowledge:read`.
`agent.get_provenance(item_id, *, version=None) -> Provenance`	Citation graph for `(item_id, version)`. Scope: `agent.knowledge:read`.
`agent.propose_candidates(request, *, idempotency_key) -> list[CandidateRecord]`	Propose candidates from an existing source. Scope: `agent.candidates:propose`.

User-tier surface (selected highlights)

The CanonClient exposes every public flycanon endpoint as a typed async method. Highlights:

# Bulk + async + replace intake
await client.submit_sources_bulk([payload_a, payload_b])
job = await client.submit_source_async(payload)
async for event in client.stream_job(job.id):
    print(event.event, event.data)
await client.replace_source(source_id, new_payload)

# Knowledge graph + diff
diff = await client.get_diff(item_id, from_version=1, to_version=2)
relations = await client.list_relations(item_id)
await client.add_relation(
    item_id,
    CreateRelationRequest(
        to_item_id=other_id,
        kind="depends_on",
    ),
)
graph = await client.get_graph(domain="compliance")
mermaid_str = await client.get_graph_mermaid(domain="compliance")

# Conversations -- same answer engine as ``answer()``, prior turns as history
conv = await client.create_conversation(CreateConversationRequest(title="t"))
turn = await client.add_turn(
    conv.id,
    CreateConversationTurnRequest(question="What about scope?"),
)
suggestions = await client.suggest_questions(SuggestRequest(question="What about scope?", answer=turn.answer))

# Streamed user-tier answer (SSE: status / hit frames, then one final)
async for frame in client.stream_answer("Summarise the scope section."):
    if frame.event == "final":
        print(frame.data["answer"], frame.data["citations"])

# Async ingest with a signed webhook + resumable progress stream
job = await client.submit_source_async(payload, callback_url="https://hooks.example.com/flycanon")
async for frame in client.stream_job(job.id):  # reconnect: stream_job(job.id, after_id=last_id)
    print(frame.event, frame.data)

# Removal + off-boarding (originals in the object store go too)
await client.delete_source(record.id)
report = await client.purge_workspace("ws-demo")  # client.workspace_id must equal "ws-demo"

# Quality scans
stale = await client.scan_stale()
report = await client.detect_conflicts(
    ConflictScanRequest(
        domain="compliance",
        min_similarity=0.85,
    )
)

# Billing + corpus inventory
summary = await client.billing_summary()
top = await client.billing_top(dimension="model", limit=5)
latency = await client.billing_latency(group_by=["model"])
snapshot = await client.stats()

Every method returns a Pydantic model (or a typed AsyncIterator for the SSE streams) -- IDE autocomplete works out of the box.

Idempotency-Key

The four agent-tier POSTs (ingest_source, query, query_stream, search, propose_candidates) require the Idempotency-Key header on the wire. The SDK exposes that as a mandatory idempotency_key= keyword argument:

await client.agent.ingest_source(spec, idempotency_key="ingest-2026-05-22-001")

Contract. The first call with a given key persists the response. Subsequent calls with the same key + same body return the same response body byte-for-byte (server-side dedup). Pass a different body with the same key and the service rejects the retry as a collision.

Local validation. The SDK rejects empty / whitespace keys with a ValueError before the request goes out so you don't pay a round-trip to discover you forgot to pass one.

await client.agent.query(req, idempotency_key="")
# ValueError: idempotency_key is required for agent-tier POSTs and must be a non-empty string

Error catalog

Every typed exception inherits from CanonAPIError, which itself inherits from CanonError. Catch the subclass to handle a specific failure; catch CanonAPIError to handle every API error generically.

Exception	HTTP	`code` (RFC 7807)
`MissingIdempotencyKey`	400	`missing_idempotency_key`
`ValidationError`	400	`invalid_request` (carries `.errors: list[FieldError]`)
`MissingAgentToken`	401	`missing_agent_token`
`InvalidAgentToken`	403	`invalid_agent_token`
`AgentTokenExpired`	403	`agent_token_expired`
`AgentWorkspaceNotInAllowlist`	403	`agent_workspace_not_in_allowlist`
`AgentScopeDenied`	403	`agent_scope_denied`
`AgentCannotMint`	403	`agent_cannot_mint`
`CanonAPIError`	any	catch-all for unrecognised codes
`CanonConnectionError`	--	network / timeout / TLS (no wire envelope)

from flycanon_sdk import (
    AgentScopeDenied,
    CanonAPIError,
    MissingIdempotencyKey,
)

try:
    await client.agent.ingest_source(spec, idempotency_key="k-001")
except MissingIdempotencyKey:
    ...
except AgentScopeDenied:
    ...
except CanonAPIError as exc:  # catch-all
    print(exc.status_code, exc.code, exc.title)

Every exception carries status_code, code, title, detail, extensions, payload, and errors. errors is a list of FieldError(code, path, message) parsed from the service's ProblemDetail.errors array -- most useful on ValidationError.

Workspace lifecycle events

flycanon emits three event types on the canon.workspaces.v1 topic. The SDK ships the typed DTOs so consumers can deserialise event payloads directly; the SDK itself does not subscribe -- wire the topic with your own EDA stack (Kafka / RabbitMQ / etc.) and use the DTOs to parse payload bodies.

from flycanon_sdk import (
    CANON_WORKSPACES_TOPIC,
    WorkspaceCreated,
    WorkspaceDeleted,
    WorkspaceUpdated,
)

assert CANON_WORKSPACES_TOPIC == "canon.workspaces.v1"


# Inside your EDA consumer:
def handle(raw_payload: dict) -> None:
    event_type = raw_payload["event_type"]
    if event_type == "workspace.created":
        evt = WorkspaceCreated.model_validate(raw_payload)
    elif event_type == "workspace.updated":
        evt = WorkspaceUpdated.model_validate(raw_payload)
    elif event_type == "workspace.deleted":
        evt = WorkspaceDeleted.model_validate(raw_payload)
    else:
        return
    print(evt.tenant_id, evt.workspace_id, evt.occurred_at)

Every event carries tenant_id, workspace_id, occurred_at, event_type, plus per-event payload fields. workspace.deleted corresponds to POST /api/v1/workspaces/{id}:close -- flycanon's terminal lifecycle transition (the row is preserved for audit; the canonical event name matches the lifecycle vocabulary other services use).

Additional resources

CHANGELOG.md -- wire-contract history per release.
openapi.json -- full route list (served by the running service at /openapi.json).
Consumers guide -- TBD, will live at flycanon/docs/consumers.md.

License

Apache-2.0 -- see LICENSE.

FilesExpand file tree

README.md

Latest commit

History