The official Klarora SDK for Python. Features an async client, a synchronous client, type hints, and Pydantic models. It covers tasks, goals, team, documents, whiteboards, and forms, and authenticates with an API key.
# pip
pip install kairos-sdk
# uv
uv add kairos-sdk
# poetry
poetry add kairos-sdk
# pipenv
pipenv install kairos-sdkKairosSync is the synchronous client for scripts and non-async frameworks:
from kairos import KairosSync
# Initialize the client
with KairosSync(api_key="kairos_sk_...") as kairos:
# List tasks
result = kairos.tasks.list(status="in_progress", limit=10)
print(f"Found {result.pagination.total} tasks")
for task in result.data:
print(f"- {task.title} ({task.status})")Kairos is the async client, for use with asyncio:
import asyncio
from kairos import Kairos
async def main():
# Initialize async client
kairos = Kairos(api_key="kairos_sk_...")
# Make async requests
result = await kairos.tasks.list(status="in_progress")
for task in result.data:
print(f"- {task.title}")
# Don't forget to close the client
await kairos.close()
# Or use as async context manager
async def main_with_context():
async with Kairos(api_key="kairos_sk_...") as kairos:
result = await kairos.tasks.list()
print(result.data)
asyncio.run(main())The examples below use the async client. With KairosSync, call the same methods without await.
The SDK authenticates with an API key (kairos_sk_...). Pass it as api_key, or omit it and set the KAIROS_API_KEY environment variable. The constructor raises ValueError if neither is set. Use kairos.me() to validate the key and see its team, scopes, and rate limits.
from kairos import Kairos
# Reads KAIROS_API_KEY from the environment
async with Kairos() as kairos:
me = await kairos.me()
print(me.team_id, me.scopes)
print(me.rate_limit_per_minute, me.rate_limit_per_hour)import os
from kairos import Kairos
kairos = Kairos(
# API key (defaults to the KAIROS_API_KEY environment variable)
api_key=os.environ["KAIROS_API_KEY"],
# Optional: Custom base URL
base_url="https://gateway.thekairos.app/v1",
# Optional: Request timeout in seconds (default: 30.0)
timeout=30.0,
# Optional: Maximum retries for rate limits and timeouts (default: 3)
max_retries=3,
)KairosSync accepts the same arguments.
# List tasks with filters
result = await kairos.tasks.list(
status="in_progress",
priority="high",
assigned_to="user-id",
goal_id="goal-id",
search="authentication",
page=1,
limit=50,
)
# Get a single task
task = await kairos.tasks.get("task-id")
print(task.title, task.status)
# Create a task
new_task = await kairos.tasks.create(
title="Implement new feature",
description="Add user authentication",
goal_id="goal-id",
priority="high",
due_date="2026-12-15",
estimated_hours=8,
)
# Update a task
updated_task = await kairos.tasks.update(
"task-id",
status="completed",
priority="low",
)
# Delete a task
await kairos.tasks.delete("task-id")
# Labels
labels = await kairos.tasks.list_labels("task-id")
await kairos.tasks.add_label("task-id", "label-id")
await kairos.tasks.remove_label("task-id", "label-id")
# Comments
comments = await kairos.tasks.list_comments("task-id", page=1, limit=10)
comment = await kairos.tasks.add_comment("task-id", content="Ready for review")
await kairos.tasks.update_comment(comment.id, content="Ready for review, tests added")
await kairos.tasks.delete_comment(comment.id)# List goals ("active", "completed" or "archived")
result = await kairos.goals.list(status="active")
# Create, get, update, delete
created = await kairos.goals.create(title="Q2 Product Launch", due_date="2026-06-30")
goal = await kairos.goals.get(created.id)
await kairos.goals.update(goal.id, status="completed")
await kairos.goals.delete(goal.id)
print(f"Goal: {goal.title}")
print(f"Progress: {goal.progress}")
# Tasks that belong to a goal
tasks = await kairos.goals.list_tasks("goal-id")
for task in tasks.data:
print(f" - {task.title}")
# Comments on a goal
comments = await kairos.goals.list_comments("goal-id")
await kairos.goals.add_comment("goal-id", content="On track")# Get team info
team = await kairos.team.get()
print(f"Team: {team.name}")
# List team members (team_id required)
result = await kairos.team.list_members(team.id)
for member in result.data:
print(f"{member.name} ({member.role})")# List documents
result = await kairos.documents.list(
type="page", # "document", "database" or "page"
search="roadmap",
)
# Get a document
doc = await kairos.documents.get("doc-id")
print(doc.title)
print(doc.content) # Rich text (Prosemirror) JSON
# Create, update, delete
created = await kairos.documents.create(title="Meeting notes")
await kairos.documents.update(created.id, title="Meeting notes (final)")
await kairos.documents.delete(created.id)# List whiteboards
result = await kairos.whiteboards.list(search="planning")
# Create, get, update, delete
board = await kairos.whiteboards.create(title="Sprint planning", is_public=False)
fetched = await kairos.whiteboards.get(board.id)
await kairos.whiteboards.update(board.id, title="Sprint 12 planning")
await kairos.whiteboards.delete(board.id)# List forms
result = await kairos.forms.list(is_active=True)
# Create a form
form = await kairos.forms.create(
name="Contact request",
fields=[
{"id": "email", "type": "email", "label": "Email", "required": True},
{"id": "message", "type": "textarea", "label": "Message", "required": False},
],
)
# Submit a response and list submissions
await kairos.forms.submit(form.id, {"email": "jane@example.com", "message": "Hello"})
submissions = await kairos.forms.list_submissions(form.id)
# Get, update, delete
await kairos.forms.get(form.id)
await kairos.forms.update(form.id, is_active=False)
await kairos.forms.delete(form.id)Every error extends KairosError. The subclasses are AuthError (401), ForbiddenError (403), NotFoundError (404), RateLimitError (429), ValidationError (400), and InternalError (5xx). All are importable from kairos.
import asyncio
from kairos import (
KairosError,
AuthError,
ForbiddenError,
NotFoundError,
ValidationError,
RateLimitError,
InternalError,
)
try:
task = await kairos.tasks.get("invalid-id")
except NotFoundError:
print("Task not found")
except ValidationError as e:
print(f"Validation error: {e.message}")
except RateLimitError as e:
print(f"Rate limited. Retry after: {e.retry_after} seconds")
await asyncio.sleep(e.retry_after)
except AuthError:
print("Authentication failed")
except KairosError as e:
# code, message, status_code and request_id are available on every error
print(f"API error: {e.code} - {e.message} ({e.status_code})")The SDK uses Pydantic models with full type hints:
from kairos import Kairos
from kairos.types import (
Task,
Goal,
Comment,
Team,
TeamMember,
Document,
Whiteboard,
Form,
FormSubmission,
PaginatedResponse,
)
kairos = Kairos(api_key="kairos_sk_...")
# All methods return typed models
task: Task = await kairos.tasks.get("id")
result: PaginatedResponse[Task] = await kairos.tasks.list()
# Access typed attributes
print(task.title) # str
print(task.status) # "to_do" | "in_progress" | "in_review" | "completed" | "cancelled"
print(task.due_date) # Optional[str]# settings.py
import os
KAIROS_API_KEY = os.environ.get("KAIROS_API_KEY")
# views.py
from django.conf import settings
from django.shortcuts import render
from kairos import KairosSync
kairos = KairosSync(api_key=settings.KAIROS_API_KEY)
def tasks_view(request):
result = kairos.tasks.list(status="in_progress")
return render(request, "tasks.html", {"tasks": result.data})from fastapi import FastAPI, Depends
from kairos import Kairos
app = FastAPI()
async def get_kairos():
async with Kairos() as client:
yield client
@app.get("/tasks")
async def get_tasks(kairos: Kairos = Depends(get_kairos)):
result = await kairos.tasks.list()
return result.data