ChatATP Studio SDK

Python SDK for building and interacting with agents created in ChatATP Studio.

• Async-first API
• Conversation lifecycle management
• Streaming support
• Fully typed

chatatp-studio

Official Python SDK for the ChatATP Studio Developer API.

Requirements

• Python 3.10+

Installation

Install the SDK core package:

pip install chatatp-studio

To enable the bundled CLI, install the optional CLI extra:

pip install "chatatp-studio[cli]"

CLI quick start

studio --help
studio auth login
studio agents --help

Quick start

import asyncio
from chatatp_studio import ChatATPClient

async def main():
    client = ChatATPClient(api_key="chatatp_sk_...")

    # Send a message — conversation lifecycle handled automatically
    result = await client.chat(
        agent_id=7,
        external_user_id="user_12345",
        message="Do you ship to Lagos?",
    )

    print(result.agent_message.content)
    # → "Yes, shipping is available."

    await client.aclose()

asyncio.run(main())

Context manager

async with ChatATPClient(api_key="chatatp_sk_...") as client:
    result = await client.chat(
        agent_id=7,
        external_user_id="user_12345",
        message="Hello!",
    )

Streaming

import sys

async for event in await client.chat_stream(
    agent_id=7,
    external_user_id="user_12345",
    message="Give me a summary of your return policy.",
):
    if event.type == "agent.response.delta":
        sys.stdout.write(event.data.get("delta", ""))
        sys.stdout.flush()
    elif event.type == "tool.execution.started":
        print(f"\n[Running tool: {event.data.get('name')}]")
    elif event.type == "tool.execution.completed":
        print(f"\n[Tool completed. Result: {event.data.get('result')}]")
    elif event.type == "error":
        print(f"\n[Error: {event.data.get('message')}]")
    elif event.type == "agent.response.completed":
        print("\nFinished!")

Resources

# Agents
page  = await client.agents.list()
agent = await client.agents.retrieve(7)

# Conversations
conv = await client.conversations.create(7, "user_12345")
page = await client.conversations.list(agent_id=7)
await client.conversations.delete(conv.id)

# Messages
history = await client.messages.list(conv.id)
reply   = await client.messages.send(conv.id, "Hello")

# Usage
usage = await client.usage.retrieve()

Knowledge bases

Knowledge bases support CRUD, document uploads, URL indexing, retrieval tests, and Agent attachments. Uploads and crawls are indexed asynchronously.

knowledge_base = await client.knowledge_bases.create(
    "Product docs",
    description="Support documentation",
    agent_id=7,  # optional: attach while creating
)

await client.knowledge_bases.upload(knowledge_base["id"], "./manual.pdf")
await client.knowledge_bases.add_url(
    knowledge_base["id"],
    "https://docs.example.com",
)

documents = await client.knowledge_bases.documents(knowledge_base["id"])
stats = await client.knowledge_bases.stats(knowledge_base["id"])
result = await client.knowledge_bases.search(
    knowledge_base["id"],
    "How do I reset my password?",
    top_k=5,
)

await client.knowledge_bases.attach(
    7,
    knowledge_base["id"],
    auto_context=True,
    max_context_chunks=5,
)

Inspect document status, chunk_count, and error_message while indexing. Use client.knowledge_bases.update() and .delete() for lifecycle management.

Memory and Agent users

Create Agent memory or associate memory with a specific Agent user:

memory = await client.memories.create(
    7,
    title="Preferred response style",
    content="The user prefers concise answers.",
    memory_type="preference",
)

user_memory = await client.memories.create(
    7,
    title="Subscription plan",
    content="The user is on the Pro plan.",
    agent_user_id=42,
    memory_type="fact",
)

memories = await client.memories.list(7, agent_user_id=42)
result = await client.memories.search(
    7,
    42,
    "What plan is the user on?",
)

users = await client.agent_users.list(7)
new_user = await client.agent_users.create(
    7,
    developer_identifier="customer_123",
    name="Jane Customer",
    metadata={"plan": "pro"},
)

Automations and schedules

Automations run scheduled builder prompts. Schedules run prompts for a specific conversation. Both resources support retrieval, updates, pause/resume, manual execution, run history, and deletion.

automation = await client.automations.create(
    agent_id=7,
    name="Daily follow-up",
    prompt_text="Review unresolved conversations and follow up.",
    schedule_type="daily",
    schedule_config={"time": "09:00"},
    timezone="UTC",
)
await client.automations.pause(automation["id"])
await client.automations.resume(automation["id"])
await client.automations.run_now(automation["id"])
runs = await client.automations.runs(automation["id"])

schedule = await client.schedules.create(
    agent_id=7,
    conversation_id=91,
    title="Renewal reminder",
    prompt_text="Remind the user about renewal.",
    schedule_type="one_time",
    schedule_config={"run_at": "2026-10-01T09:00:00Z"},
)
await client.schedules.run_now(schedule["id"])
schedule_runs = await client.schedules.runs(schedule["id"])

CLI resources

Install the CLI extra with pip install "chatatp-studio[cli]":

studio kb documents upload <knowledge-base-id> --file ./manual.pdf
studio kb search <knowledge-base-id> --query "reset password"
studio memory search <agent-id> --query "response style"
studio automations run-now <automation-id>
studio automations runs <automation-id>
studio schedules pause <schedule-id>
studio schedules runs <schedule-id>

Error handling

from chatatp_studio import NotFoundError, RateLimitError

try:
    await client.agents.retrieve(999)
except NotFoundError:
    print("Not found")
except RateLimitError:
    print("Rate limited")

License

MIT