Cominty Python SDK

Official async Python client for the Cominty managed agent chat API.

Start a conversation with an agent, stream its progress live, and manage threads — with a small, fully-typed surface that's the same on Python 3.9 through 3.13.

importasynciofromcominty_sdkimportAsyncComintyasyncdefmain() ->None:
asyncwithAsyncCominty() asclient: # reads COMINTY_API_KEY + COMINTY_USER_IDrun=awaitclient.chat.start(agent_id="__cominty_agents::agent.chat",
message="What is Cominty?")
print(awaitrun.text())
asyncio.run(main())

• Async-first, built on httpx.
• Fully typed — ships py.typed; strict-checked with pyright. Pydantic models everywhere.
• One handle for streaming and awaiting — iterate a run for live progress events, or just await run.text() for the final answer.
• Fail-fast validation — bad parameters raise locally, before any request.
• Typed errors — every failure is a ComintyError subclass.

---

Requirements

• Python 3.9+
• A Cominty API key and your user id (see Authentication)

Installation

pip install cominty-sdk
# or
uv add cominty-sdk

Authentication

You need two things, both from platform.cominty.ai:

1. API key → platform.cominty.ai/api-keys (shown once — copy it).
2. Your user id → avatar (top right) → Profile. It looks like user_31HPTBuBvX20xlQNAbvxjOxPbKB.

The user id identifies the end user every request acts on behalf of. It's set once on the client (or via COMINTY_USER_ID) and applied to every call.

The simplest setup is environment variables:

export COMINTY_API_KEY="<your API key>"export COMINTY_USER_ID="user_..."

asyncwithAsyncCominty() asclient: # picks both up from the environment
...

…or pass them explicitly (explicit arguments win over the environment):

client=AsyncCominty(api_token="<your API key>", user_id="user_...")

A malformed user_id is rejected at construction, not as a server error later.

Picking an agent

Every chat call takes an agent_id. Browse your agents and copy an id at platform.cominty.ai/agents — they look like __cominty_agents::agent.chat.

Quick start

Every conversation starts with chat.start, which returns a run — a handle to the assistant's in-progress reply. From there, pick the style you need.

Just get the answer

run=awaitclient.chat.start(agent_id=AGENT_ID, message="Give me one fun fact.")
print(awaitrun.text()) # blocks until the agent finishes

await run.result() gives the full Message (status, files, structured output, questions). text() is shorthand for result().content.

Stream progress events

Iterating a run yields progress events only — tool calls, LLM steps, the result event — as they happen. The finished reply is captured for you.

fromcominty_sdkimporteventsrun=awaitclient.chat.start(agent_id=AGENT_ID, message="Research X and summarize.")
asyncforeventinrun:
ifisinstance(event, events.ToolCall):
print(f"tool {event.data.name} -> {event.status}")
elifisinstance(event, events.LlmStep):
print(f"llm {event.data.description}")
elifisinstance(event, events.Result):
print(f"cost {event.data.cost.total}")
print("FINAL:", awaitrun.text()) # available after the stream drains

A run's stream is single-use: iterate it or await its result — the result is cached, so calling text()/result() after iterating is free.

Continue the conversation

chat.send(thread_id, ...) is the mirror of start for an existing thread: same arguments, same streamable run. The agent keeps the thread's context.

first=awaitclient.chat.start(agent_id=AGENT_ID, message="Pick a language.")
awaitfirst.text()
second=awaitclient.chat.send(
first.thread.id, agent_id=AGENT_ID, message="Now show hello-world in it.",
)
print(awaitsecond.text())

Answer the agent's questions

When an agent needs more input, it ends its turn with clarifying questions (a prompt plus suggested options) instead of a final answer. Read them, then answer with a normal follow-up:

run=awaitclient.chat.start(agent_id=AGENT_ID, message="Book me a room.")
awaitrun.text()
forqinawaitrun.questions():
print(q.prompt, q.options)
# answer = the chosen option (or free text)reply=awaitclient.chat.send(run.thread.id, agent_id=AGENT_ID, message="Tomorrow 10am")
print(awaitreply.text())

Manage threads

client.threads is scoped to the client's user_id automatically.

# List & search the user's conversations (summaries — no messages)fortinawaitclient.threads.list(limit=20):
print(t.created_at, t.name, t.id)
awaitclient.threads.list(terms=["invoice"]) # free-text searchawaitclient.threads.list(limit=10, page=1) # paginate (zero-based)# Load one thread's full historythread=awaitclient.threads.get(thread_id)
print(len(thread.messages))
# Partial update — only the fields you pass change (returns a ThreadSummary)awaitclient.threads.update(thread_id, name="Renamed", starred=True)
# Archive (soft-delete)awaitclient.threads.archive(thread_id)

Examples

Runnable scripts for each scenario live in examples/:

Script	Shows
01_stream_events.py	Stream progress events live
02_await_result.py	Fire and await the final answer
03_follow_up.py	Continue in the same thread
04_answer_questions.py	Read & answer agent questions
05_list_threads.py	List and search threads
06_manage_thread.py	Get, rename/star, archive
07_custom_agent.py	Call a custom managed agent (your own model + instructions)
08_mcp_linear.py	Custom agent pulls live context from an MCP server (Linear)

They render colored, aligned output with rich, which ships in the dev extras:

uv sync --all-extras --dev # installs rich (or: pip install rich)export COMINTY_API_KEY=... COMINTY_USER_ID=user_...
python examples/01_stream_events.py

Message parameters

Both chat.start and chat.send accept:

Argument	Type	Notes
agent_id	str	Required. The agent to run.
message	str	Required. The user's message (max 30,000 chars).
name	str	start only — names the new thread.
file_ids	list[str]	Attach previously-uploaded files (max 5).
source_ids	list[int]	Restrict retrieval to specific knowledge sources.
document_ids	list[str]	Restrict retrieval to specific documents.
disabled_tools	list[str]	Turn tools off: "web", "company_documents", "mcp:<server>", or "mcp:*" for all MCP.

Invalid values raise InvalidParamsbefore any request is sent.

Configuration

Argument	Env var	Default
api_token	COMINTY_API_KEY	— (required)
user_id	COMINTY_USER_ID	— (required)
base_url	COMINTY_BASE_URL	https://ds.cominty.com
timeout	—	60 (seconds)

Resolution order for each option: explicit argument → environment variable → default. The SDK does not auto-load .env; export the vars or load the file yourself (see .env.example).

Error handling

Every error is a subclass of ComintyError:

fromcominty_sdkimport (
ComintyError, # base — catch-allAPIError, # any 4xx/5xx; carries .status_code and a typed .error bodyAuthError, # 401PermissionError, # 403NotFoundError, # 404ConflictError, # 409RateLimitError, # 429 — exposes .reset_atServerError, # 5xxAPIConnectionError, # network failure / timeout, no responseStreamInterrupted, # server shut down mid-stream — carries the .partial MessageInvalidParams, # client-side validation failed — .errors lists each problemSDKError, # unexpected SDK-internal condition
)
try:
run=awaitclient.chat.start(agent_id=AGENT_ID, message="hi")
print(awaitrun.text())
exceptRateLimitErrorase:
print(f"slow down — retry after {e.reset_at}")
exceptAPIErrorase:
print(f"API error {e.status_code}: {e.error}")

Development

uv sync --all-extras --dev
uv run pytest # tests
uv run ruff check .# lint
uv run pyright # type-check (strict)

Integration tests are opt-in (they hit the real API):

COMINTY_API_KEY=... COMINTY_USER_ID=... uv run pytest -m integration

See AGENTS.md for coding conventions (typing, versioning, models).

Releasing

Publishing to PyPI uses Trusted Publishing (OIDC) — no tokens stored in GitHub — and is triggered by publishing a GitHub Release (.github/workflows/release.yml). The published version comes from pyproject.toml, so the tag is cosmetic; keep them in sync.

# 1. bump the version in pyproject.toml# 2. commit on main and push# 3. create the release — this tags and triggers the publish
gh release create v0.4.0 --title "v0.4.0" --generate-notes
# pre-release rehearsal: gh release create v0.4.0rc1 --prerelease --generate-notes

A local rehearsal to TestPyPI is available via uv run invoke publish-test.

License

MIT