copass-core

Core client SDK for the Copass platform. Python mirror of @copass/core — shared foundation for every Python Copass adapter.

Install

pip install copass-core

Requires httpx>=0.27. Python ≥ 3.10.

Quickstart

import asyncio
from copass_core import CopassClient, ApiKeyAuth

async def main():
    client = CopassClient(auth=ApiKeyAuth(key="olk_..."))

    # Retrieval
    menu = await client.retrieval.discover(
        sandbox_id="sb_...",
        query="How does auth work?",
    )
    print(menu["items"])

asyncio.run(main())

Auth options

from copass_core import CopassClient, ApiKeyAuth, BearerAuth, ProviderAuth

# Long-lived API key (olk_ prefix)
CopassClient(auth=ApiKeyAuth(key="olk_..."))

# Raw Bearer JWT (caller owns refresh)
CopassClient(auth=BearerAuth(token="eyJ..."))

# Custom AuthProvider implementation
class MyProvider:
    async def get_session(self):
        from copass_core import SessionContext
        return SessionContext(access_token=await _mint_token())

CopassClient(auth=ProviderAuth(provider=MyProvider()))

Available resources

Full resource surface matching @copass/core:

client = CopassClient(auth=ApiKeyAuth(key="olk_..."))

# Narrow retrieval tools
await client.retrieval.discover(sandbox_id, query="...")
await client.retrieval.interpret(sandbox_id, query="...", items=[...])
await client.retrieval.search(sandbox_id, query="...")

# Storage layer
await client.sandboxes.create(name="...", owner_id="...")
await client.sources.register(sandbox_id, provider="custom", name="...")
await client.ingest.text_in_sandbox(sandbox_id, text="...")
await client.projects.create(sandbox_id, name="...")
await client.vault.store(sandbox_id, "key/path", b"bytes")

# Knowledge graph
await client.entities.search(sandbox_id, q="auth")

# Account
await client.users.get_profile()
await client.api_keys.create(name="ci")
await client.usage.get_balance()

# Higher-order — ephemeral data source wrapping agent conversation
window = await client.context_window.create(sandbox_id=sandbox_id)
await window.add_turn(ChatMessage(role="user", content="..."))
# Pass directly to retrieval for window-aware calls:
await client.retrieval.search(sandbox_id, query="...", window=window)

Reaching your sandbox

Compute sessions ship with a per-session reverse-proxy gateway (ADR 0026) so you can hit any port inside the sandbox over HTTPS without managing your own tunnel. client.compute.create_session / get_session / list_sessions return ComputeSession instances that expose proxy_url, websocket_url, and fetch:

session = await client.compute.create_session(
    sandbox_id, template="copass-hermes-py311", timeout_seconds=600,
)
# Hit port 3000 inside the sandbox via the public gateway.
resp = await session.fetch(3000, "/api/v1/health")
print(resp.status_code, await resp.aread())
print(session.proxy_url(3000, "/dashboard"))      # https://...
print(session.websocket_url(8080, "/ws"))         # wss://...
await client.compute.stop_session(sandbox_id, session.session_id)

fetch is a thin passthrough — bearer auth is added for you, but body, headers, method, and timeout flow through httpx untouched (no JSON serialization, no retries, no error normalization). Pass "" for the bare per-port URL or a string starting with / for a sub-path.

Conversation metadata: speaker, participants, timestamp

Every ingestion path accepts optional metadata that travels alongside the content on the envelope. Set them when the data is conversation-shaped so the platform can attribute and order content correctly.

Direct API

await client.ingest.text(
    text="Hey Alice, did you finish the report?",
    source_type="conversation",
    speaker="Bob",
    participants=["Alice", "Bob"],
    occurred_at="2026-05-08T15:30:00Z",
)

Through a ContextWindow

For a chat-style agent, set the participant roster once at window construction; it forwards on every add_turn call. Set ChatMessage.name when you want a richer speaker than the role-derived default ("User" / "Assistant"):

window = await client.context_window.create(
    sandbox_id=sandbox_id,
    participants=["User", "Alice"],   # default roster, applied on every turn
)

# Role-only — speaker derived as capitalized role.
await window.add_turn(ChatMessage(role="user", content="Hey Alice…"))

# Named participant — `name` overrides role-derived speaker.
await window.add_turn(
    ChatMessage(role="user", content="…thanks!", name="Bob"),
)

# Per-call override — use when the roster shifts mid-conversation.
await window.add_turn(
    ChatMessage(role="assistant", content="…"),
    participants=["User", "Alice", "Bob", "Carol"],
)

Through a BaseDataSource subclass

class SlackChannelSource(BaseDataSource):
    async def push_message(self, msg):
        await self.push(
            msg.text,
            source_type="conversation",
            speaker=msg.author_name,
            participants=msg.channel.member_names,
            occurred_at=msg.posted_at_iso,
        )

Existing callers that don't pass any of these fields keep working — all four are optional.

v0.2 scope

Shipped in v0.2:

• Full resource surface (12 resource classes, all public paths).
• ContextWindow + ContextWindowResource.
• BaseDataSource + ensure_data_source for custom driver subclasses.
• HttpClient raw-body / raw-response support (enables vault blob I/O).

Deferred to v0.3:

• Crypto primitives (HKDF, AES-GCM, session tokens, DEK).
• Supabase OTP auth provider (requires crypto).
• BearerAuth(encryption_key=...) currently stores the key but doesn't derive a session token — works when the server doesn't demand one.

Open a PR with a scoped addition if you need those sooner.

License

MIT.