context-studio-sdk 0.1.2

Python SDK for Context Studio - context enrichment platform

Context Studio SDK

Python SDK for Context Studio — a context enrichment platform for AI applications.

Context Studio lets you build AI apps that remember users. The SDK provides a clean, typed interface to retrieve personal context, extract and store personal data, and integrate all of this seamlessly into LangGraph workflows.

---

Table of Contents

• Installation
• Quick Start
• Configuration
• Async & Sync Usage
• API Reference
  • Health Check
  • Runtimes
  • Runtime Operations
  • Ontology
  • API Keys
• LangGraph Integration
  • Enriched LLM Runnable
  • Context Studio Tools
  • Standalone Node Functions
  • Playground Compare
  • Protocol Adapter
• Error Handling
• Examples
• Development
• License

---

Installation

pip install context-studio-sdk

With LangGraph support (recommended for agentic workflows):

pip install context-studio-sdk[langgraph]

For development:

pip install context-studio-sdk[dev]

---

Quick Start

from context_studio_sdk import ContextStudio

cs = ContextStudio(
    base_url="https://api.context.studio",
    api_key="cs_xxxxx",
    project_uuid="proj_xxxxx",
    runtime_uuid="rt_xxxxx",
)

# Retrieve context-enriched text for a user query
result = cs.runtimes.operations.retrieve_sync(
    "What should I do this weekend?",
    user_key="alex",
)
print(result.enriched_text)
print(result.labels)

# Ingest personal data from text
ingest = cs.runtimes.operations.ingest_sync(
    sentences=["My name is Alex and I live in Tokyo"],
    user_key="alex",
)
print(f"Task: {ingest.task_id} — {ingest.status}")

# Clean up
cs.close()

Async usage (recommended for production):

import asyncio
from context_studio_sdk import ContextStudio

async def main():
    async with ContextStudio(
        base_url="https://api.context.studio",
        api_key="cs_xxxxx",
        project_uuid="proj_xxxxx",
        runtime_uuid="rt_xxxxx",
    ) as cs:
        result = await cs.runtimes.operations.retrieve(
            "What should I do this weekend?",
            user_key="alex",
        )
        print(result.enriched_text)

asyncio.run(main())

---

Configuration

The ContextStudio client accepts the following parameters:

Parameter	Type	Default	Description
base_url	str	—	Required. Base URL of the Context Studio deployment (e.g. "https://api.context.studio"). Trailing slashes are stripped automatically. Must start with http:// or https://.
api_key	str	—	Required. API key for authentication. Sent as the X-API-Key header on every request.
project_uuid	str	—	Required. Project UUID. Sent as the project-uuid header to scope all operations to a specific project.
runtime_uuid	str | None	None	Optional runtime instance UUID. When set, runtime operations (retrieve, ingest, contextualize, etc.) are scoped to this specific instance. Required for cs.runtimes.operations.* methods.
timeout	float	30.0	HTTP request timeout in seconds. Applies to all requests made through this client.
max_retries	int	3	Maximum number of automatic retries on transport-level failures (connection errors).

Authentication

Every HTTP request includes two headers:

• X-API-Key: <your_api_key> — authenticates the caller.
• project-uuid: <your_project_uuid> — scopes the request to the target project.

These are configured once at client initialization and applied automatically.

---

Async & Sync Usage

All SDK methods come in two variants:

• Async (default): Use with await in async contexts. These are the primary method names (e.g., retrieve(), create(), list()).
• Sync: Append _sync to the method name (e.g., retrieve_sync(), create_sync(), list_sync()). These use a blocking HTTP client internally — no asyncio.run() wrapper needed, so they work inside Jupyter notebooks and other environments with an existing event loop.

Async pattern:

async with ContextStudio(base_url="...", api_key="...", project_uuid="...") as cs:
    result = await cs.runtimes.operations.retrieve("hello", user_key="alex")

Sync pattern:

cs = ContextStudio(base_url="...", api_key="...", project_uuid="...")
try:
    result = cs.runtimes.operations.retrieve_sync("hello", user_key="alex")
finally:
    cs.close()

The client supports both sync and async context managers:

# Async context manager — closes async client on exit
async with ContextStudio(...) as cs:
    ...

# Sync context manager — closes sync client on exit
with ContextStudio(...) as cs:
    ...

---

API Reference

Health Check

Check the status of the Context Studio deployment.

# Async
health = await cs.health()

# Sync
health = cs.health_sync()

print(health.status)      # "healthy"
print(health.central_db)  # "connected"

Returns a HealthResponse with service and database status. This endpoint does not require authentication.

---

Runtimes

Manage runtime instances. Accessed via cs.runtimes.

Method	Signature	Description
create	(ontology_uuid, *, title=None, description=None, settings=None) → UppRuntimeResponse	Create and start a new runtime backed by a published ontology.
list	(*, status=None) → list[UppRuntimeResponse]	List runtimes, optionally filtered by UppRuntimeStatus.RUNNING or UppRuntimeStatus.STOPPED.
get	(runtime_uuid) → UppRuntimeResponse	Retrieve a single runtime by UUID.
update	(runtime_uuid, *, title=None, description=None, settings=None) → UppRuntimeResponse	Update runtime metadata and/or settings. Only provided fields are updated.
start	(runtime_uuid) → UppRuntimeResponse	Start a stopped runtime.
stop	(runtime_uuid) → UppRuntimeResponse	Stop a running runtime.
delete	(runtime_uuid, *, preserve_db=False) → UppRuntimeResponse	Delete a runtime. Set preserve_db=True to keep the runtime database on disk.
activate	(runtime_uuid) → UppRuntimeResponse	Set a runtime as the active instance for the project. Only one can be active at a time.

Example:

from context_studio_sdk.models import UppRuntimeStatus

# Create a runtime
runtime = await cs.runtimes.create(
    ontology_uuid="onto_xxxxx",
    title="Production v1",
    description="Main production instance",
)

# List running runtimes
running = await cs.runtimes.list(status=UppRuntimeStatus.RUNNING)

# Activate a runtime for the project
await cs.runtimes.activate(runtime.uuid)

# Runtime lifecycle
await cs.runtimes.stop(runtime.uuid)
await cs.runtimes.start(runtime.uuid)
await cs.runtimes.delete(runtime.uuid)

---

Runtime Operations

Core data operations — retrieve context, ingest data, contextualize, manage tasks and events. Accessed via cs.runtimes.operations.

Note: All operations require runtime_uuid to be configured on the client.

Retrieve

Method	Signature	Description
retrieve	(text, user_key="default") → RetrieveResponse	Classify text and retrieve relevant context for a user. Returns enriched text, matched events, and labels.

result = await cs.runtimes.operations.retrieve(
    "I love hiking in the mountains",
    user_key="alex",
)
print(result.enriched_text)
print(result.labels)
print(result.events)

Ingest

Method	Signature	Description
ingest	(sentences, user_key="default") → IngestResponse	Submit a batch of sentences for background extraction. Returns a task ID for polling.
poll_task	(task_id, *, interval=2.0, timeout=300.0) → TaskResponse	Poll a task until it reaches a terminal state (completed or failed). Async only.
get_task	(task_id) → TaskResponse	Retrieve the current status of a pipeline task.
list_tasks	(user_key, status=None) → list[TaskResponse]	List pipeline tasks for a user, optionally filtered by status.

# Submit a batch of sentences for extraction
ingest_result = await cs.runtimes.operations.ingest(
    sentences=[
        "My name is Alex and I'm 28 years old",
        "I live in Tokyo and work at Acme Corp",
        "I love hiking and photography",
    ],
    user_key="alex",
)
print(f"Task ID: {ingest_result.task_id}")

# Poll until completion
task = await cs.runtimes.operations.poll_task(
    ingest_result.task_id,
    interval=2.0,
    timeout=60.0,
)
print(f"Status: {task.status}")

Contextualize

Method	Signature	Description
contextualize	(text, user_key="default") → ContextualizeResponse	Retrieve relevant context and schedule background ingest in a single call. Returns existing events and a task ID for the ingest.

This is the most commonly used operation — it combines retrieval and ingestion in one step:

result = await cs.runtimes.operations.contextualize(
    "I just moved to Tokyo and love Japanese food",
    user_key="alex",
)
print(f"Existing events: {result.events}")
print(f"Ingest task: {result.task_id}")

Events & User Keys

Method	Signature	Description
list_events	(user_key, source_message_ids=None, label=None) → list[EventResponse]	List extracted events for a user, optionally filtered by source message IDs or label.
list_labels	() → list[dict]	List all labels available in the runtime's ontology.
list_user_keys	() → list[str]	List all user keys that have ingested data.

# List stored events for a user
events = await cs.runtimes.operations.list_events("alex")
for event in events:
    print(f"  {event.value} — labels: {event.labels}")

# Discover which users have data
user_keys = await cs.runtimes.operations.list_user_keys()
print(f"Users: {user_keys}")

---

Ontology

Full ontology management — CRUD operations, labels, publishing, validation, training, and configuration. Accessed via cs.ontology.

Ontology CRUD

Method	Signature	Description
create	(body: OntologyCreate) → OntologyResponse	Create a new ontology.
list	() → list[OntologyResponse]	List all ontologies in the project.
get	(ontology_uuid) → OntologyResponse	Retrieve a single ontology by UUID.
get_sdk_default	() → OntologyExport	Retrieve the SDK-default ontology export.
update	(ontology_uuid, body: OntologyUpdate) → OntologyResponse	Update an existing ontology.
delete	(ontology_uuid) → OntologyResponse	Delete an ontology.

Labels

Method	Signature	Description
create_label	(ontology_uuid, body: OntologyLabelCreate) → OntologyLabelResponse	Add a new label to an ontology.
list_labels	(ontology_uuid) → list[OntologyLabelResponse]	List all labels in an ontology.
get_label	(ontology_uuid, label_id) → OntologyLabelResponse	Get a specific label by ID.
update_label	(ontology_uuid, label_id, body: OntologyLabelUpdate) → OntologyLabelResponse	Update a label.
delete_label	(ontology_uuid, label_id) → OntologyLabelResponse	Delete a label.

Lifecycle Operations

Method	Signature	Description
publish	(ontology_uuid) → MessageResponse	Publish an ontology, making it available for runtimes.
validate	(ontology_uuid) → ValidationResponse	Validate an ontology's structure and labels.
generate_schema	(ontology_uuid) → MessageResponse	Trigger JSON schema generation for the extraction pipeline.
train_classifier	(ontology_uuid, body: TrainingSettings) → MessageResponse	Start background classifier training (synthetic data generation + fine-tuning).
get_classifier_status	(ontology_uuid) → ClassifierStatusResponse	Check the current classifier training status and progress.

Export / Import

Method	Signature	Description
export_ontology	(ontology_uuid) → OntologyExport	Export an ontology (metadata + labels) as a portable JSON structure.
import_ontology	(body: OntologyExport) → OntologyResponse	Import an ontology from a previously exported structure.

Configuration

Method	Signature	Description
get_exclusion_config	(ontology_uuid) → dict	Get the exclusion configuration (mutually exclusive labels).
update_exclusion_config	(ontology_uuid, body: ExclusionConfigUpdate) → OntologyResponse	Update the exclusion configuration.
get_bundles	(ontology_uuid) → dict	Get bundle definitions (grouped labels for batch operations).
update_bundles	(ontology_uuid, body: BundlesUpdate) → OntologyResponse	Update bundle definitions.
get_remap_config	(ontology_uuid) → dict	Get the label remap configuration (aliases/redirects).
update_remap_config	(ontology_uuid, body: RemapConfigUpdate) → OntologyResponse	Update the remap configuration.

Example:

from context_studio_sdk.models import OntologyCreate, OntologyLabelCreate, TrainingSettings

# Create an ontology
ontology = await cs.ontology.create(OntologyCreate(
    name="User Preferences",
    description="Labels for tracking user preferences and personal data",
))

# Add labels
await cs.ontology.create_label(ontology.uuid, OntologyLabelCreate(
    name="food_preference",
    display_name="Food Preference",
    description="User's food preferences, dietary restrictions, favorite cuisines",
    category="what",
))

# Validate and publish
validation = await cs.ontology.validate(ontology.uuid)
if validation.valid:
    await cs.ontology.publish(ontology.uuid)

# Train a classifier
await cs.ontology.train_classifier(
    ontology.uuid,
    TrainingSettings(model="default", synthetic_data_count=100),
)

---

API Keys

Manage API keys for the project. Accessed via cs.api_keys.

Method	Signature	Description
create	(body: ApiKeyCreate) → ApiKeyCreateResponse	Create a new API key. The full token is only returned at creation time.
list	() → list[ApiKeyResponse]	List all API keys (tokens are not included — only prefixes).
get	(key_uuid) → ApiKeyResponse	Retrieve a single API key by UUID.
update	(key_uuid, body: ApiKeyUpdate) → ApiKeyResponse	Update an API key (name, active status, expiration, IP whitelist).
delete	(key_uuid) → None	Delete an API key (returns 204 No Content).

Example:

from context_studio_sdk.models import ApiKeyCreate

# Create a new API key
key = await cs.api_keys.create(ApiKeyCreate(name="Production Key"))
print(f"Token (save this!): {key.token}")  # Only shown once

# List existing keys
keys = await cs.api_keys.list()
for k in keys:
    print(f"  {k.name}: {k.prefix}...")

---

LangGraph Integration

The LangGraph integration provides pre-built graph components that inject Context Studio's context enrichment into your LangGraph workflows.

Requires the langgraph extra:

pip install context-studio-sdk[langgraph]

The integration is accessed via cs.langgraph and includes:

• create_runnable() — A complete enriched LLM graph (quick-start helper).
• create_tools() — LangChain tools wrapping SDK operations.
• Standalone node functions — contextualize_node for custom graphs (recommended for production).
• playground — Side-by-side comparison of responses with and without context.

---

Enriched LLM Runnable

cs.langgraph.create_runnable() builds a compiled LangGraph StateGraph that automatically enriches user messages with personal context before passing them to an LLM.

Graph Architecture

Without tools:

START → contextualize → llm → END

With tools:

START → contextualize → llm → conditional(tools/END) → tools → llm (loop)

The contextualize node:

1. Finds the last HumanMessage in the conversation.
2. Calls the Context Studio /contextualize endpoint to retrieve relevant context and schedule background ingestion.
3. Injects the context as a SystemMessage before the LLM call.

Parameters

Parameter	Type	Default	Description
llm	BaseChatModel	—	Required. Any LangChain chat model.
tools	Sequence[BaseTool] | None	None	Optional LangChain tools for tool-calling.
checkpointer	Any | None	None	Optional LangGraph checkpointer for state persistence.
context_format	str	"flat"	Format for context injection.

Complete Example

import asyncio
from context_studio_sdk import ContextStudio
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage

async def main():
    cs = ContextStudio(
        base_url="https://api.context.studio",
        api_key="cs_xxxxx",
        project_uuid="proj_xxxxx",
        runtime_uuid="rt_xxxxx",
    )
    llm = ChatAnthropic(model="claude-sonnet-4-5-20250514")

    # Build the enriched runnable
    runnable = cs.langgraph.create_runnable(llm)

    # Invoke — context is automatically retrieved and injected
    result = await runnable.ainvoke({
        "messages": [HumanMessage(content="What should I do this weekend?")],
        "user_key": "alex",
    })

    # The LLM response includes context-aware recommendations
    print(result["messages"][-1].content)

    await cs.aclose()

asyncio.run(main())

Direct Import

You can also import the graph builder directly:

from context_studio_sdk.langgraph.graph import create_context_studio_runnable

runnable = create_context_studio_runnable(cs, llm)

---

Context Studio Tools

cs.langgraph.create_tools() creates LangChain tools that wrap SDK operations. These tools can be bound to an LLM or passed to the enriched runnable.

The user_key is captured at creation time so the tools only require task-specific inputs.

Available Tools

Tool Name	Input	Description
get_events	(none)	Retrieve stored personal-data events for the current user as a JSON array.
ingest_data	text: str	Extract personal data from text and ingest it for the current user.

Example

from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage

llm = ChatAnthropic(model="claude-sonnet-4-5-20250514")

# Create tools bound to a specific user
tools = cs.langgraph.create_tools(user_key="alex")

# Option 1: Bind directly to LLM
llm_with_tools = llm.bind_tools(tools)

# Option 2: Use with the enriched runnable (tool-calling loop)
runnable = cs.langgraph.create_runnable(llm, tools=tools)
result = await runnable.ainvoke({
    "messages": [HumanMessage(content="Remember that my favorite food is sushi")],
    "user_key": "alex",
})

---

Standalone Node Functions

For maximum flexibility, the SDK exposes a standalone async node function that you can integrate into custom LangGraph graphs. This is the recommended approach for production use — compose this node alongside your own in a single flat graph instead of nesting compiled graphs.

from functools import partial
from context_studio_sdk.langgraph.nodes import (
    ContextStudioState,
    contextualize_node,
)
from langgraph.graph import END, START, StateGraph

# Bind the client at graph-build time
bound_contextualize = partial(contextualize_node, client=cs)

# Build a custom graph
graph = StateGraph(ContextStudioState)
graph.add_node("contextualize", bound_contextualize)
graph.add_node("llm", your_llm_node)
graph.add_edge(START, "contextualize")
graph.add_edge("contextualize", "llm")
graph.add_edge("llm", END)
compiled = graph.compile()

Available Node Functions

Function	Adds to State	Description
contextualize_node(state, *, client)	messages: [SystemMessage]	Calls the /contextualize endpoint (retrieve context + schedule background ingest). Injects context as a SystemMessage.

The node expects state with messages: list and user_key: str keys (matching ContextStudioState).

---

Playground Compare

The PlaygroundCompare client runs a user query through both a plain LLM and an enriched (Context Studio) LLM and returns both responses side-by-side. This is useful for demonstrating the value of context enrichment.

Accessed via cs.langgraph.playground.

# Requires runtime_uuid configured on the client
cs = ContextStudio(
    base_url="https://api.context.studio",
    api_key="cs_xxxxx",
    project_uuid="proj_xxxxx",
    runtime_uuid="rt_xxxxx",
)

# Remote comparison (server-side)
result = await cs.langgraph.playground.compare(
    query="What's my favorite food?",
    user_key="alex",
    model="claude-sonnet-4-5-20250514",  # optional
)

print("Without context:", result.raw_response)
print("With context:", result.enriched_response)

# Local comparison (client-side, requires langgraph extra)
from langchain_anthropic import ChatAnthropic
llm = ChatAnthropic(model="claude-sonnet-4-5-20250514")

result = await cs.langgraph.playground.compare_local(
    llm=llm,
    query="What's my favorite food?",
    user_key="alex",
)

Method	Signature	Description
compare	(query, user_key, model=None) → PlaygroundCompareResponse	Run a side-by-side comparison on the server (async).
compare_sync	(query, user_key, model=None) → PlaygroundCompareResponse	Run a side-by-side comparison on the server (sync).
compare_local	(llm, query, user_key) → PlaygroundCompareResponse	Run a side-by-side comparison locally using LangGraph (async).

---

Protocol Adapter

For advanced use cases, the ContextStudioHTTPAdapter provides a simplified async interface:

adapter = cs.langgraph.get_adapter()

# Simplified async methods
enriched_text = await adapter.retrieve_enriched_text("hiking plans", "alex")
events = await adapter.get_events("alex")
task_info = await adapter.ingest_sentences(["I love sushi"], "alex")

---

Error Handling

All SDK exceptions inherit from ContextStudioError, so you can catch broad or specific errors:

ContextStudioError (base)
├── AuthenticationError    # HTTP 401 — invalid or missing API key
├── NotFoundError          # HTTP 404 — resource not found
├── ValidationError        # HTTP 422 — invalid request body
├── RateLimitError         # HTTP 429 — too many requests
├── ServerError            # HTTP 5xx — server-side failure
├── TimeoutError           # Request timed out
└── ConnectionError        # Cannot connect to the server

Usage Patterns

from context_studio_sdk import (
    ContextStudioError,
    AuthenticationError,
    NotFoundError,
    RateLimitError,
    TimeoutError,
    ConnectionError,
)

# Catch a specific error
try:
    result = await cs.runtimes.operations.retrieve("hello", user_key="alex")
except AuthenticationError:
    print("Invalid API key — check your credentials")
except NotFoundError:
    print("Resource not found — is the runtime running?")
except RateLimitError:
    print("Rate limited — slow down or upgrade your plan")

# Catch all SDK errors
try:
    result = await cs.runtimes.operations.retrieve("hello", user_key="alex")
except ContextStudioError as e:
    print(f"SDK error: {e}")
    print(f"Status code: {e.status_code}")
    print(f"Detail: {e.detail}")

# Handle connection issues
try:
    health = await cs.health()
except ConnectionError:
    print("Cannot reach the server — check base_url and network")
except TimeoutError:
    print("Request timed out — the server may be under load")

All exceptions include:

• status_code (int | None): The HTTP status code that triggered the error.
• detail (str | None): Human-readable detail message from the server.

---

Examples

The examples/ directory contains runnable scripts demonstrating common workflows:

• basic_usage.py — SDK initialization, health check, runtime management, context retrieval (async + sync).
• enrich_and_ingest.py — Full retrieve → ingest → poll → list events pipeline.
• langgraph_integration.py — Enriched LLM runnables, tool-calling graphs, playground comparison.

---

Development

Setup

git clone <repository-url>
cd context-studio-sdk
pip install -e ".[dev,langgraph]"

Run Tests

pytest

Lint

ruff check src/ tests/
ruff format src/ tests/

Project Structure

context-studio-sdk/
├── src/
│   └── context_studio_sdk/
│       ├── __init__.py          # Package exports
│       ├── _http.py             # HTTP transport layer (httpx)
│       ├── _version.py          # Version string
│       ├── client.py            # ContextStudio main client
│       ├── config.py            # SDKConfig (Pydantic)
│       ├── exceptions.py        # Exception hierarchy
│       ├── models/              # Pydantic request/response models
│       │   ├── __init__.py
│       │   ├── api_keys.py
│       │   ├── events.py
│       │   ├── health.py
│       │   ├── runtimes.py
│       │   ├── ontology.py
│       │   └── sdk_execution.py
│       ├── api/                 # API resource modules
│       │   ├── __init__.py
│       │   ├── api_keys.py
│       │   ├── runtimes.py
│       │   ├── runtime_operations.py
│       │   └── ontology.py
│       └── langgraph/           # LangGraph integration
│           ├── __init__.py
│           ├── graph.py         # Graph builders
│           ├── nodes.py         # Standalone node functions
│           ├── playground.py    # Playground comparison
│           └── protocol.py      # HTTP adapter
├── tests/
├── examples/
├── docs/
├── pyproject.toml
├── CHANGELOG.md
└── README.md

---

License

MIT License. See LICENSE for details.