kopernica 0.2.0

Python SDK for the Kopernica platform — agents, memory, and sessions as a service.

Kopernica Python SDK

Python client for the Kopernica platform — a 6-agent AI pipeline with persistent memory and managed sessions, delivered as an API.

Kopernica runs a parallel multi-agent workflow (Coordinator, Context, Sentiment, Query, Voice, Analytics) against your users, maintains per-user memory across conversations, and returns human-context-aware responses.

---

Installation

pip install kopernica

Requirements: Python 3.10+

---

Quick Start

Sync

from kopernica import KopernicaClient

kop = KopernicaClient(api_key="kop_your_key_here")

response = kop.orchestrate(
    "What should I focus on for my upcoming interview?",
    end_user_id="user-42",
)

print(response.content)          # The AI-generated response
print(response.total_latency_ms) # Pipeline processing time
print(response.agent_traces)     # Per-agent execution details

Async

from kopernica import AsyncKopernicaClient

async with AsyncKopernicaClient(api_key="kop_your_key_here") as kop:
    response = await kop.orchestrate(
        "What should I focus on for my upcoming interview?",
        end_user_id="user-42",
    )
    print(response.content)

---

Authentication

All requests require an API key passed via the X-API-Key header. The SDK handles this automatically.

kop = KopernicaClient(
    api_key="kop_...",
    base_url="https://your-deployment.run.app/api/v1",  # optional override
    timeout=60.0,                                         # request timeout in seconds
    max_retries=3,                                        # retries on 429/5xx (default 3)
)

Both clients support context managers:

# Sync
with KopernicaClient(api_key="kop_...") as kop:
    resp = kop.orchestrate("Hello", end_user_id="user-1")

# Async
async with AsyncKopernicaClient(api_key="kop_...") as kop:
    resp = await kop.orchestrate("Hello", end_user_id="user-1")

Getting an API Key

Use the bootstrap endpoint (requires a Firebase superuser account):

curl -X POST https://your-api.run.app/api/v1/platform/bootstrap \
  -H "Authorization: Bearer <firebase_token>" \
  -H "Content-Type: application/json" \
  -d '{
    "tenant": {
      "name": "My Company",
      "slug": "my-company",
      "contact_email": "[e-mail]"
    },
    "first_key_name": "production"
  }'

The response includes a raw_key field (starting with kop_) — this is returned only once. Store it securely.

---

Core Concepts

end_user_id

Every call requires an end_user_id — this is your user's identifier. Kopernica scopes all data (sessions, memory, agent context) to this ID within your tenant. Use whatever ID system you already have (database ID, auth UID, etc.).

Allowed characters: alphanumeric, dots, underscores, @ signs, and hyphens. Other characters will be rejected with a 422 error.

session_id

Sessions group conversation turns together. You can either:

• Let Kopernica auto-create sessions — omit session_id from orchestrate calls and a new session is created and persisted automatically
• Manage sessions explicitly — call kop.sessions.create() and pass the returned ID

Auto-created sessions appear in kop.sessions.list() just like explicitly created ones.

Context Buckets

Memory is organized into context buckets: career, networking, job_search, learning, productivity, general. The system auto-infers the bucket from message content, or you can override it.

---

API Reference

Orchestrate

kop.orchestrate() / await kop.orchestrate()

Send a message through the full 6-agent pipeline and get the complete response.

response = kop.orchestrate(
    "How should I prepare for a product manager interview?",
    end_user_id="user-42",
    session_id="sess-abc",                # optional — auto-created if omitted
    conversation_history=[                 # optional — for stateless usage
        {"role": "user", "content": "I have an interview next week"},
        {"role": "assistant", "content": "That's exciting! What role?"},
    ],
    user_preferences={                     # optional — controls response style
        "response_style": "detailed",
        "language": "en",
    },
    context_bucket="career",              # optional — auto-inferred if omitted
)

Returns: OrchestrateResponse

Field	Type	Description
content	str	The generated response text
session_id	str	Session used (auto-created or existing)
end_user_id	str	Echo of the end-user ID
agent_traces	list[AgentTrace]	Per-agent execution details
context_data	dict	Output from the Context Agent (memories, history)
sentiment_data	dict	Output from the Sentiment Agent (emotion, tone)
query_data	dict	Output from the Query Agent (intent, entities)
total_latency_ms	int	Total pipeline processing time
success	bool	Whether the pipeline succeeded
error	str | None	Error message if failed

kop.orchestrate_stream() / kop.orchestrate_stream() (async)

Stream the pipeline as Server-Sent Events. Useful for showing real-time progress in a UI.

Sync:

for event in kop.orchestrate_stream("Tell me about my goals", end_user_id="user-42"):
    if event.type == "text_chunk":
        print(event.data.get("chunk", ""), end="", flush=True)

Async:

async for event in kop.orchestrate_stream("Tell me about my goals", end_user_id="user-42"):
    if event.type == "text_chunk":
        print(event.data.get("chunk", ""), end="", flush=True)

Yields: StreamEvent

Event Type	Description
agent_start	An agent has begun executing
agent_complete	An agent finished (includes latency, status)
text_chunk	One chunk of the Voice Agent's streaming response
done	Pipeline complete (includes total_latency_ms)
error	Unrecoverable failure

---

Memory

Kopernica maintains a multi-lane memory system (Mnemosyne) for each end-user. Memory is automatically extracted from conversations, but you can also write and read it directly.

kop.memory.write()

Write a conversation turn into memory. The system extracts facts, preferences, and context automatically. Returns immediately with a task_id — processing happens async.

result = kop.memory.write(
    end_user_id="user-42",
    session_id="sess-abc",
    user_message="I'm preparing for a PM interview at Google next Tuesday",
    assistant_message="Great! Let's work on your preparation strategy...",
    context_bucket="career",  # optional
)

print(result["task_id"])  # Use to poll write status

You can poll the write status to confirm it completed:

GET /api/v1/platform/memory/write-status/{task_id}
# Returns: {"task_id": "...", "status": "processing" | "completed" | "failed", "error": "..."}

kop.memory.retrieve()

Retrieve the structured memory pack for a user. Returns memories organized by lane.

pack = kop.memory.retrieve(
    end_user_id="user-42",
    context_bucket="career",
    query="interview preparation",  # optional — improves relevance
)

print(pack.working_set)      # Active goals and next actions
print(pack.facts_prefs)      # Durable user profile details
print(pack.long_term)        # Time-aware personal history
print(pack.mid_term)         # Recent event/idea cues
print(pack.short_term)       # Immediate session context
print(pack.general_overlay)  # Cross-context memories
print(pack.context_bucket)   # The bucket used for retrieval

Returns: MemoryPack

Lane	Description	Example
working_set	Currently active goals and next actions	"Prepare for Google PM interview"
facts_prefs	Durable profile details and preferences	"Prefers detailed explanations"
long_term	Time-aware personal history with temporal bounds	"Worked at Meta 2020-2023"
mid_term	Compact event/idea cues (subject to decay)	"Mentioned salary expectations"
short_term	Immediate session continuity snapshots	Last few turns context
general_overlay	Cross-context memories that apply everywhere	"User is based in San Francisco"

kop.memory.list()

List individual memory entries with filtering and pagination.

entries = kop.memory.list(
    end_user_id="user-42",
    bucket="career",           # optional filter
    lane="facts_prefs",        # optional filter
    page=1,
    page_size=20,
)

for entry in entries:
    print(f"[{entry.memory_type}] {entry.topic_key}: {entry.value}")
    print(f"  confidence={entry.confidence}, status={entry.status}")

Returns: list[MemoryEntry]

Deleting memory entries

Memory entries can be soft-deleted via the REST API:

DELETE /api/v1/platform/memory/entries/{entry_id}
# Requires X-API-Key header. Verifies tenant ownership. Returns 204.

---

Sessions

Sessions group conversation turns and provide continuity. Memory consolidation runs automatically when a session ends.

kop.sessions.create()

session = kop.sessions.create(
    end_user_id="user-42",
    channel="api",  # "api", "web", "mobile"
)

print(session.id)      # Use this in orchestrate() calls
print(session.status)  # "active"

kop.sessions.list()

sessions = kop.sessions.list(
    end_user_id="user-42",
    page=1,
    page_size=10,
)

for s in sessions:
    print(f"{s.id} | {s.status} | {s.current_topic} | {s.created_at}")

kop.sessions.get_turns()

turns = kop.sessions.get_turns("sess-abc")

for turn in turns:
    print(f"[{turn.role}] {turn.content}")
    if turn.intent:
        print(f"  intent={turn.intent}, sentiment={turn.sentiment}")

kop.sessions.end()

End a session. This triggers memory consolidation — the system extracts session-level facts and stores them in long-term memory.

kop.sessions.end("sess-abc")

---

Complete Example: Conversational Agent

from kopernica import KopernicaClient

kop = KopernicaClient(api_key="kop_your_key_here")

# Create a session for the user
session = kop.sessions.create(end_user_id="user-42")

# Multi-turn conversation
messages = [
    "I'm looking for advice on transitioning to product management",
    "I have 5 years of engineering experience at a startup",
    "What skills should I focus on developing?",
]

history = []
for msg in messages:
    resp = kop.orchestrate(
        msg,
        end_user_id="user-42",
        session_id=session.id,
        conversation_history=history,
    )

    print(f"User: {msg}")
    print(f"AI:   {resp.content}\n")

    history.append({"role": "user", "content": msg})
    history.append({"role": "assistant", "content": resp.content})

    kop.memory.write(
        end_user_id="user-42",
        session_id=session.id,
        user_message=msg,
        assistant_message=resp.content,
    )

# End session — triggers memory consolidation
kop.sessions.end(session.id)

# Later... retrieve what the system remembers
pack = kop.memory.retrieve(end_user_id="user-42", context_bucket="career")
print("Working set:", pack.working_set)
print("Facts:", pack.facts_prefs)

---

Complete Example: Async FastAPI Integration

from fastapi import FastAPI
from kopernica import AsyncKopernicaClient

app = FastAPI()
kop = AsyncKopernicaClient(api_key="kop_your_key_here")


@app.post("/chat")
async def chat(user_id: str, message: str):
    resp = await kop.orchestrate(message, end_user_id=user_id)
    return {"reply": resp.content, "latency_ms": resp.total_latency_ms}


@app.on_event("shutdown")
async def shutdown():
    await kop.close()

---

Complete Example: Streaming with Progress UI

from kopernica import KopernicaClient

kop = KopernicaClient(api_key="kop_your_key_here")

print("Agents processing...\n")

full_response = ""
for event in kop.orchestrate_stream(
    "Analyze my interview readiness and suggest a preparation plan",
    end_user_id="user-42",
):
    match event.type:
        case "agent_start":
            print(f"  [{event.data['agent']}] started")
        case "agent_complete":
            status = "done" if event.data.get("success") else "failed"
            print(f"  [{event.data['agent']}] {status} ({event.data.get('latency_ms')}ms)")
        case "text_chunk":
            chunk = event.data.get("chunk", "")
            full_response += chunk
            print(chunk, end="", flush=True)
        case "done":
            print(f"\n\nPipeline complete in {event.data.get('total_latency_ms')}ms")
        case "error":
            print(f"\nError: {event.data.get('error')}")

---

Error Handling & Retries

The SDK automatically retries on transient failures (429, 502, 503, 504 and network errors) with exponential backoff. Default: 3 retries with 0.5s/1s/2s delays.

# Disable retries
kop = KopernicaClient(api_key="kop_...", max_retries=0)

# More retries for unreliable networks
kop = KopernicaClient(api_key="kop_...", max_retries=5)

For non-retryable errors, the SDK raises KopernicaError:

from kopernica.client import KopernicaError

try:
    resp = kop.orchestrate("Hello", end_user_id="user-1")
except KopernicaError as e:
    print(f"API error {e.status_code}: {e.detail}")

Status Code	Meaning	Retried?
401	Invalid or missing API key	No
403	Key expired/deactivated, tenant inactive, scope denied	No
404	Resource not found (session, memory entry)	No
422	Validation error (bad request body, invalid end_user_id)	No
429	Rate limit exceeded	Yes
502	Bad gateway	Yes
503	Service unavailable	Yes
504	Gateway timeout	Yes

Rate Limiting

The platform enforces per-tenant rate limits (default: 60 requests/minute). The SDK surfaces rate limit headers in the error:

try:
    resp = kop.orchestrate("Hello", end_user_id="user-1")
except KopernicaError as e:
    if e.status_code == 429:
        print("Rate limited — SDK will retry automatically")

Response headers on every request:

• X-RateLimit-Limit — your tenant's RPM limit
• X-RateLimit-Remaining — requests left in current window
• X-RateLimit-Reset — Unix timestamp when the window resets

---

Agent Pipeline Architecture

When you call orchestrate(), the message flows through 6 specialized agents:

                          +------------------+
                          |   Coordinator    |   Analyzes request, routes to agents
                          +--------+---------+
                                   |
                    +--------------+--------------+
                    |              |              |
              +-----+----+  +----+------+  +----+-----+
              | Context  |  | Sentiment |  |  Query   |    Run in PARALLEL
              |  Agent   |  |   Agent   |  |  Agent   |
              +-----+----+  +----+------+  +----+-----+
                    |              |              |
                    +--------------+--------------+
                                   |
                          +--------+---------+
                          |   Voice Agent    |   Generates response using
                          |                  |   enriched context from all agents
                          +--------+---------+
                                   |
                          +--------+---------+
                          | Analytics Agent  |   Logs async (non-blocking)
                          +------------------+

Agent	What it does	Runs
Coordinator	Routes request, synthesizes all results	First
Context	Retrieves memories, session history, user preferences	Parallel
Sentiment	Detects emotion, escalation risk, recommends tone	Parallel
Query	Classifies intent, extracts entities, enhances query	Parallel
Voice	Generates the final response using enriched context	After parallel agents
Analytics	Logs conversation data for analysis	Async (non-blocking)

---

Memory System Architecture

Kopernica uses Mnemosyne, a multi-lane memory system with automatic decay:

Memory Lanes

Lane	Purpose	Lifecycle
short_term	Immediate session continuity	Session-scoped
mid_term	Compact event/idea cues	Core (30d) > Dream (90d) > Forgotten (180d)
working_set	Active goals and next actions	Dynamic, updated per turn
facts_prefs	Durable profile details	Persistent, high confidence
long_term	Time-aware personal history	Persistent with temporal bounds

Context Buckets

Memory is organized into situational domains so the agent only retrieves what's relevant:

Bucket	Topics
career	Jobs, interviews, skills, resume
networking	Connections, events, outreach
job_search	Applications, companies, offers
learning	Courses, certifications, skills
productivity	Goals, tasks, habits
general	Everything else

---

Configuration

Parameter	Default	Description
api_key	required	Your kop_... API key
base_url	Cloud Run URL	API base URL
timeout	60.0	Request timeout in seconds
max_retries	3	Retries on 429/5xx and network errors

---

Development

# Clone and install in dev mode
cd sdk/python
pip install -e ".[dev]"

# Run tests
pytest tests/

---

Changelog

0.1.0

• Initial release
• Sync client (KopernicaClient) and async client (AsyncKopernicaClient)
• Orchestrate (full response + SSE streaming)
• Memory (write with task tracking, retrieve, list)
• Sessions (create, list, get turns, end)
• Automatic retries with exponential backoff on transient errors
• Rate limit headers surfaced on every response