Quickstart - Cominty Docs

Every scenario below has a runnable script in the SDK’s examples/ directory. Each snippet assumes you are inside an AsyncCominty context and have an AGENT_ID:

import asyncio
from cominty_sdk import AsyncCominty, events

AGENT_ID = "__cominty_agents::agent.chat"

async def main():
    async with AsyncCominty() as client:          # COMINTY_API_KEY + COMINTY_USER_ID
        ...  # the snippets below go here

asyncio.run(main())

Run the examples from a checkout of the repo:

uv sync --all-extras --dev    # examples render with `rich`
python examples/01_stream_events.py

New to the SDK? Read Overview first for install, auth, and the core concepts (client, resources, runs, tools).

Stream progress events

Watch the agent work (tool calls, LLM steps, the result) as it happens. Iterating a run yields progress events only; the finished reply is captured for you.

run = await client.chat.start(agent_id=AGENT_ID, message="Research X and summarize.")
async for event in run:
    if isinstance(event, events.ToolCall):
        print("tool", event.data.name, event.status)
    elif isinstance(event, events.LlmStep):
        print("llm ", event.data.description)
    elif isinstance(event, events.Result):
        print("cost", event.data.cost.total)
print(await run.text())

▶ examples/01_stream_events.py

Fire and await the answer

No event handling. Just the final reply.

run = await client.chat.start(agent_id=AGENT_ID, message="Give me one fun fact.")
print(await run.text())            # blocks until done
reply = await run.result()         # full Message (status, files, ...)

▶ examples/02_await_result.py

Continue the conversation

chat.send posts a follow-up in the same thread; the agent keeps context.

A thread handles one message at a time. Read the previous reply to the end (await run.text()), or cancel it, before you send the next message. A message you start and never read stays pending, and chat.send then fails with APIError (400, Cannot process multiple messages at the same time). See One message at a time.

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

▶ examples/03_follow_up.py

Answer questions from the agent

When the agent needs input, it ends its turn with questions (a prompt plus options) instead of a final answer. Answer with a normal follow-up.

run = await client.chat.start(agent_id=AGENT_ID, message="Book me a room.")
await run.text()
for q in await run.questions():
    print(q.prompt, q.options)
reply = await client.chat.send(run.thread.id, agent_id=AGENT_ID, message="Tomorrow 10am")
print(await reply.text())

▶ examples/04_answer_questions.py

Cap the tool rounds

max_steps limits how many tool rounds the agent runs for one message. When the cap is reached, the agent recaps and asks whether to continue. The message still ends with status="success". The cap applies to that message only, so pass it again on a follow-up. Sub-agents have their own counter. See Limiting agent steps.

run = await client.chat.start(
    agent_id=AGENT_ID,
    message="Research X, then write it up.",
    max_steps=5,
)
await run.text()
reply = await client.chat.send(
    run.thread.id,
    agent_id=AGENT_ID,
    message="Yes, continue.",
    max_steps=10,
)
print(await reply.text())

▶ examples/10_max_steps.py

Stop a message

chat.cancel stops an assistant message that is still being generated. It returns the message with status set to cancelled. It is not idempotent: cancelling a message that already finished raises APIError (400). A “stop” button should treat that error as “too late”.

from cominty_sdk import APIError

run = await client.chat.start(
    agent_id=AGENT_ID, message="Write a very detailed 3000 word market report."
)

try:
    message = await client.chat.cancel(run.message_id)
    print(message.status)  # MessageStatus.cancelled
except APIError as error:
    print(error.status_code, error)  # 400 Message is not live

A cancelled message frees the thread: you can send the next message right away, without reading the cancelled message’s events. ▶ examples/12_cancel_message.py

Export a message as a document

chat.export renders a finished message to a PDF or a DOCX and returns the bytes. The server sends no file name, so pick one when you write the file.

from pathlib import Path

run = await client.chat.start(
    agent_id=AGENT_ID,
    message="Write a short product brief about a note-taking app, in Markdown.",
)
reply = await run.result()

pdf = await client.chat.export(reply.id, format="pdf")  # or "docx"
Path("brief.pdf").write_bytes(pdf)

A message with no content, for example one that is still running, raises APIError (400). Any format other than "pdf" or "docx" raises InvalidParams before a request is sent. ▶ examples/13_export_message.py

Give the agent files

chat.upload_file registers a file. Its id goes in file_ids on chat.start or chat.send, at most 5 files per message. Pass a pathlib.Path and the name and mimetype are inferred. Pass bytes with a filename. A plain str is rejected: wrap a path in Path(...).

from pathlib import Path

report = await client.chat.upload_file(Path("q3-report.pdf"))
data = await client.chat.upload_file(b"a,b\n1,2\n", filename="data.csv")

run = await client.chat.start(
    agent_id=AGENT_ID,
    message="Compare these two files.",
    file_ids=[report.id, data.id],
)
reply = await run.result()

# Read a file back, including the ones the agent produced.
content = await client.chat.download_file(report.id)
for file in reply.files:
    # Keep only the last path segment: file names are not under your control.
    Path(Path(file.name).name).write_bytes(await client.chat.download_file(file.id))

A file is at most 500 MiB, held in memory. A bigger one raises InvalidParams before anything is sent. See the reference for the limits and errors. ▶ examples/14_upload_files.py

List and search threads

for t in await client.threads.list(limit=20):
    print(t.created_at, t.name, t.id)
await client.threads.list(terms=["invoice"])   # free-text search
await client.threads.list(limit=10, page=1)     # paginate (zero-based)

▶ examples/05_list_threads.py

Manage a thread

thread = await client.threads.get(thread_id)                          # full history
await client.threads.update(thread_id, name="Renamed", starred=True)  # partial → ThreadSummary
await client.threads.archive(thread_id)                               # soft-delete

▶ examples/06_manage_thread.py

Store and recall a memory file

client.memory stores files in a namespace you choose. The same API key shares those files across user_id values. update is partial, and its required version guards against overwriting a concurrent change. A stale one raises ConflictError. To use the files in a conversation, pass a MemoryCapability in thread_capabilities to chat.start. The concepts are in Memory namespaces.

from cominty_sdk import (
    AgentCapabilities,
    ConflictError,
    MemoryCapability,
)

namespace = "support-bot"

created = await client.memory.create(
    path="preferences/tone.md",
    namespace=namespace,
    purpose="writing style",
    content="Keep it casual.",
)

summaries = await client.memory.list(namespace=namespace)  # no content
fetched = await client.memory.get("preferences/tone.md", namespace=namespace)

updated = await client.memory.update(
    "preferences/tone.md",
    namespace=namespace,
    version=fetched.version,
    content="Keep it upbeat.",
)

try:
    await client.memory.update(
        "preferences/tone.md",
        namespace=namespace,
        version=fetched.version,
        content="stale",
    )
except ConflictError:
    print("version was stale. Re-fetch and retry")

await client.memory.delete("preferences/tone.md", namespace=namespace)

run = await client.chat.start(
    agent_id=AGENT_ID,
    message="Use the tone file.",
    thread_capabilities=AgentCapabilities(
        memory=MemoryCapability(activation="always", namespace=namespace),
    ),
)

▶ examples/09_memory.py

Control what the agent can use

Capabilities decide which tools the agent can use: web search, document search, MCP connections, skills, media generation, and memory. Set them for a whole thread with thread_capabilities (on chat.start only), or for one message with message_scope (on chat.start and chat.send).

from cominty_sdk import (
    AgentCapabilities,
    MessageScope,
    WebCapability,
    WebScope,
)

run = await client.chat.start(
    agent_id=AGENT_ID,
    message="What is the refund window for annual plans?",
    thread_capabilities=AgentCapabilities(
        web=WebCapability(activation="on_request"),  # off until a message asks
    ),
)
print(await run.text())

reply = await client.chat.send(
    run.thread.id,
    agent_id=AGENT_ID,
    message="Check a competitor's pricing online.",
    message_scope=MessageScope(web=WebScope(enabled=True)),  # this message only
)
print(await reply.text())

▶ examples/11_capabilities.py See Capabilities for the model, and Use cases for recipes.

Use a custom managed agent

A custom agent (its own model, failover order, and instructions, configured at platform.cominty.ai → Agents) is just another agent_id. Pass it to any chat call. This example feeds a jargon-heavy incident report to an agent instructed to reply with a clean, non-technical French executive summary.

from cominty_sdk import IndexedDocumentsScope, McpScope, MessageScope, WebScope

run = await client.chat.start(
    agent_id=CUSTOM_AGENT_ID,
    message=TECHNICAL_INPUT,
    message_scope=MessageScope(  # pure transformation: no tools
        web=WebScope(enabled=False),
        indexed_documents=IndexedDocumentsScope(enabled=False),
        mcp=McpScope(enabled=False),
    ),
)
print(await run.text())   # bullet-point French briefing for a C-level reader

▶ examples/07_custom_agent.py

Pull live context from an MCP server

MCP servers connected to an agent on the platform (for example Linear) work with no SDK change. Capabilities are on by default, and a message scope turns off what you don’t want. Here the custom agent reads the current Linear sprint and reports it in its French C-level voice. Linear tool calls appear in the stream as ToolCall events.

from cominty_sdk import IndexedDocumentsScope, MessageScope, WebScope

run = await client.chat.start(
    agent_id=CUSTOM_AGENT_ID,
    message="Using Linear, list the current-sprint tasks and summarize for leadership.",
    message_scope=MessageScope(  # keep MCP (Linear) on, cut the rest
        web=WebScope(enabled=False),
        indexed_documents=IndexedDocumentsScope(enabled=False),
    ),
)
print(await run.text())

▶ examples/08_mcp_linear.py

List the MCP integrations

mcp.list returns the integrations your API key can use. A server’s name is the value McpCapability.connections stores. By default it keeps only the servers that are turned on.

servers = await client.mcp.list()
for server in servers:
    print(server.name, server.title, server.requires_oauth_connect)

everything = await client.mcp.list(enabled=None)   # every server, on or off

requires_oauth_connect is True when someone must connect the integration on the platform first. If the manifest cannot be loaded, the SDK raises ServerError. ▶ examples/19_mcp_servers.py

Create a custom agent

Create an agent from code, with its own instructions and capabilities. The agent is owned by the user who created your API key. Chat with it like any other agent.

from cominty_sdk import AgentCapabilities, McpCapability, WebCapability

agent = await client.agents.create(
    name="Integration helper",
    instructions="Use your connected integration when it helps. Be brief.",
    scope="organization",
    capabilities=AgentCapabilities(
        mcp=McpCapability(activation="always", connections=[servers[0].name]),
        web=WebCapability(activation="never"),
    ),
)
run = await client.chat.start(
    agent_id=agent.id,
    message="Which integration can you use? Answer in one sentence.",
)
print(await run.text())

agent = await client.agents.update(agent.id, name="Helper v2", description=None)

update is partial: None clears description, instructions, and capabilities, and a capabilities object is merged capability by capability. See Custom agents. ▶ examples/16_agents.py

Choose the agent’s models

agents.list_models gives the model ids. set_models replaces the agent’s chain: the first model answers, and the next ones take over when it fails.

models = await client.agents.list_models()
agent = await client.agents.set_models(
    agent.id, model_ids=[models[1].id, models[0].id]
)
print([model.name for model in agent.models])

await client.agents.delete(agent.id)

On a built-in agent, set_models changes the chain of the whole organization.

Give an agent a skill

A skill is a folder with a SKILL.md. skills.create zips the folder and uploads it. Give it to an agent with SkillsCapability.

from pathlib import Path
from cominty_sdk import AgentCapabilities, SkillsCapability

skill = await client.skills.create(Path("release-notes"), display_name="Release notes")
agent = await client.agents.create(
    name="Greeter",
    capabilities=AgentCapabilities(
        skills=SkillsCapability(activation="always", include=[skill.id]),
    ),
)
run = await client.chat.start(agent_id=agent.id, message="Give me the QA greeting.")
print(await run.text())

See Skills for the folder layout and the versions. ▶ examples/17_skills.py

Search your own files

sources.create turns local files into an indexed source in one call. Wait for it to be synced, then give its id to an agent. The chat must run with the key owner’s user_id, which is agent.owner.

from pathlib import Path
from cominty_sdk import (
    AgentCapabilities,
    AsyncCominty,
    IndexedDocumentsCapability,
)

source = await client.sources.create(Path("docs"), name="Product docs")
source = await client.sources.wait_synced(source.pid, timeout=420, interval=3)

agent = await client.agents.create(
    name="Docs assistant",
    instructions="Answer only from the company documents. Cite the documents you used.",
    capabilities=AgentCapabilities(
        indexed_documents=IndexedDocumentsCapability(
            activation="always", source_ids=[source.id]
        ),
    ),
)
async with AsyncCominty(user_id=agent.owner) as owner_client:
    run = await owner_client.chat.start(agent_id=agent.id, message="What is the code word?")
    message = await run.result()
print(message.content)

source.pid (a string) goes to every client.sources method. source.id (an integer) goes to source_ids. See Sources. ▶ examples/15_sources.py

Resolve a citation

An answer cites documents as <cite document_id="2698#10778" .../>. documents.get resolves that id to the source and the file.

import re

cites = re.findall(r'<cite document_id="([^"]+)"', message.content)
document = await client.documents.get(cites[0])
print(document.document_name, document.source_pid, document.document_uri)

files = await client.sources.list_files(document.source_pid)
content = await client.sources.download_file(files[0].id)   # bytes

An unknown id raises NotFoundError. There is no call that lists documents.

Handle errors

from cominty_sdk import RateLimitError, APIError

try:
    run = await client.chat.start(agent_id=AGENT_ID, message="hi")
    print(await run.text())
except RateLimitError as e:
    print(e)                       # clear, scope-aware message
    print(e.scope, e.retry_after, e.reset_at)
except APIError as e:
    print(e.status_code, e.detail)

See Exceptions for the full hierarchy.

​Stream progress events

​Fire and await the answer

​Continue the conversation

​Answer questions from the agent

​Cap the tool rounds

​Stop a message

​Export a message as a document

​Give the agent files

​List and search threads

​Manage a thread

​Store and recall a memory file

​Control what the agent can use

​Use a custom managed agent

​Pull live context from an MCP server

​List the MCP integrations

​Create a custom agent

​Choose the agent’s models

​Give an agent a skill

​Search your own files

​Resolve a citation

​Handle errors

Stream progress events

Fire and await the answer

Continue the conversation

Answer questions from the agent

Cap the tool rounds

Stop a message

Export a message as a document

Give the agent files

List and search threads

Manage a thread

Store and recall a memory file

Control what the agent can use

Use a custom managed agent

Pull live context from an MCP server

List the MCP integrations

Create a custom agent

Choose the agent’s models

Give an agent a skill

Search your own files

Resolve a citation

Handle errors