Metadata-Version: 2.4
Name: context212-sdk
Version: 0.1.0
Summary: Type-annotated Python client for the Context212 API
Author: Context212
Author-email: Context212 <devops@quiztr.com>
License-Expression: MIT
Requires-Dist: httpx>=0.20,<0.29
Requires-Dist: attrs>=21.3.0
Requires-Dist: python-dateutil>=2.8.0,<3
Requires-Python: >=3.11
Project-URL: Repository, https://github.com/context212/ctx212-python-sdk
Project-URL: Issues, https://github.com/context212/ctx212-python-sdk/issues
Description-Content-Type: text/markdown

# context212-sdk

[![PyPI](https://img.shields.io/pypi/v/context212-sdk)](https://pypi.org/project/context212-sdk/)
[![Docs](https://img.shields.io/badge/docs-docs.context212.com-blue)](https://docs.context212.com)
[![Python](https://img.shields.io/badge/python-3.11%20%7C%203.12%20%7C%203.13-blue)](https://github.com/context212/ctx212-python-sdk)

Seamlessly integrate state-of-the-art document intelligence directly into your software.

## What is Context212?

Context212 is a document-intelligence platform: upload your documents, then query them
with grounded **ask** and **search** actions, and process documents on the fly with
**parse** and **extract** for specific, standalone operations.

This SDK wraps the Context212 API. Create an account and get an API key on
[console.context212.com](https://console.context212.com) 🚀

## Contents

- [Quick start](#quick-start)
- [Configuration](#configuration)
- [Primary verbs](#primary-verbs)
- [Workspaces](#workspaces)
- [Files & ingestion](#files--ingestion)
- [Async jobs & polling](#async-jobs--polling)
- [Tags](#tags)
- [Error handling](#error-handling)
- [Agent frameworks](#agent-frameworks)
- [Develop](#develop)

## Quick start

Install:

```bash
pip install context212-sdk
# or: uv add context212-sdk
```

Set your API key in your environment:

```bash
export CONTEXT212_API_KEY="..."
```

Get your first result:

```python
import os

from context212_sdk import AuthenticatedClient
from context212_sdk.api.ask import ask
from context212_sdk.api.files import files_create
from context212_sdk.api.search import search
from context212_sdk.api.workspaces import create_workspace
from context212_sdk.models import (
    AskRequest,
    BodyFilesCreate,
    SearchRequest,
    StandardWorkspaceCreateRequest,
)
from context212_sdk.types import File

client = AuthenticatedClient(
    base_url="https://api.context212.com",
    token=os.environ["CONTEXT212_API_KEY"],
)

# Create a workspace and upload a PDF
ws = create_workspace.sync(
    client=client, body=StandardWorkspaceCreateRequest(name="Docs")
)
with open("report.pdf", "rb") as f:
    files_create.sync(
        client=client,
        body=BodyFilesCreate(file=File(payload=f), workspace_id=ws.id),
    )

# Search: retrieve the most relevant passages, scoped to that workspace
hits = search.sync(
    client=client, body=SearchRequest(query="Q4 revenues", workspace_id=[ws.id])
)
for r in hits.results:
    print(r.score, r.content)

# Ask: single-turn RAG, grounded answer plus the sources used
answer = ask.sync(
    client=client, body=AskRequest(query="What were Q4 revenues?", workspace_id=[ws.id])
)
print(answer.answer)
```

Every endpoint function comes in two flavors: `sync` / `sync_detailed` for blocking
code and `asyncio` / `asyncio_detailed` for async code.

## Configuration

`base_url` is the host only — the SDK appends `/api/v1/...` paths itself. Pass
`headers`, `timeout`, `verify_ssl`, `follow_redirects`, or `httpx_args` to tune the
underlying httpx client:

```python
import httpx

from context212_sdk import AuthenticatedClient

client = AuthenticatedClient(
    base_url="http://localhost:8000",  # self-hosted / staging / local
    token="...",
    timeout=httpx.Timeout(600.0, connect=5.0),
)
```

The client works as a context manager (`with client:`) when you want explicit
connection lifecycle, or bare for short-lived scripts.

## Primary verbs

Four actions live under `context212_sdk.api`. `ask` and `search` query your
**indexed** documents — scope them with `workspace_id`, `tag_id`, or `file_id`.
`parse` and `extract` process a document **on the fly**, no indexing required. Full
reference at [docs.context212.com](https://docs.context212.com).

### `ask`: single-turn RAG

Retrieves the most relevant chunks and has an LLM answer your question grounded in
them, returning the answer **plus the sources it used**. Add `stream=True` for
Server-Sent Events.

```python
from context212_sdk.api.ask import ask
from context212_sdk.models import AskRequest

resp = ask.sync(
    client=client,
    body=AskRequest(
        query="What were Q4 revenues?", workspace_id=[ws.id], max_results=5
    ),
)
print(resp.answer)
```

### `search`: retrieval only, no generation

Hybrid retrieval returning ranked chunks with scores and source metadata, but no
generated answer. Use it to feed context into your own pipeline/LLM, build custom
ranking, or surface sources to users.

```python
from context212_sdk.api.search import search
from context212_sdk.models import RelevanceScoringEnum, SearchRequest

resp = search.sync(
    client=client,
    body=SearchRequest(
        query="termination clause",
        tag_id=[7],
        relevance_scoring=RelevanceScoringEnum.SCORING_ONLY,
    ),
)
for r in resp.results:
    print(r.score, r.content)
```

`relevance_scoring` tunes the scoring step (applies to `ask` too):

- omitted: score and drop chunks below the quality threshold
- `SCORING_ONLY`: score every candidate, return them all
- `NONE`: skip scoring; lowest latency

### `parse`: document → Markdown

One-off conversion of a document into structured per-page Markdown, without storing
it in your index. Pass a publicly accessible `document` URL:

```python
from context212_sdk.api.parse import parse
from context212_sdk.models import ParseJsonRequest

job = parse.sync(
    client=client, body=ParseJsonRequest(document="https://example.com/report.pdf")
)
```

Large documents can exceed the sync timeout — queue them with
`options={"async": True}` and poll (see [Async jobs & polling](#async-jobs--polling)).

### `extract`: schema-guided structured data

Pull typed fields out of a document: you describe the shape as a JSON Schema and get
back data matching it. `document` takes a public URL, or target a file you already
ingested with `file_id`:

```python
from context212_sdk.api.extract import extract
from context212_sdk.models import ExtractRequest, ExtractRequestSchema

resp = extract.sync(
    client=client,
    body=ExtractRequest(
        document="https://example.com/invoice.pdf",
        schema=ExtractRequestSchema.from_dict(
            {
                "type": "object",
                "properties": {
                    "total": {
                        "type": "number",
                        "description": "Invoice total, as printed.",
                    },
                    "currency": {
                        "type": ["string", "null"],
                        "description": "ISO 4217 code.",
                    },
                },
                "required": ["total"],
            }
        ),
    ),
)
```

Give every field a meaningful `description` — descriptions steer the model and
materially improve extraction quality. Treat them as instructions, not documentation.

## Workspaces

Workspaces are the containers your documents live in; retrieval scopes to them.

```python
from context212_sdk.api.workspaces import (
    create_workspace,
    delete_workspace,
    get_workspace,
    list_workspaces,
    update_workspace,
)
from context212_sdk.models import StandardWorkspaceCreateRequest

ws = create_workspace.sync(
    client=client,
    body=StandardWorkspaceCreateRequest(name="Legal", description="Contracts & NDAs"),
)

all_ws = list_workspaces.sync(client=client)
one = get_workspace.sync(client=client, id=ws.id)

update_workspace.sync(client=client, id=ws.id, body=...)
delete_workspace.sync(client=client, id=ws.id)
```

## Files & ingestion

Uploading a file into a workspace *is* the ingestion — a `File` carries a processing
`status` you can poll with `files_retrieve` until it reaches `embedded`. Once embedded,
it's retrievable by `ask`/`search`.

```python
import time

from context212_sdk.api.files import (
    files_create,
    files_destroy,
    files_list,
    files_retrieve,
)
from context212_sdk.models import BodyFilesCreate
from context212_sdk.types import File

with open("report.pdf", "rb") as f:
    doc = files_create.sync(
        client=client,
        body=BodyFilesCreate(
            file=File(payload=f), workspace_id=ws.id, title="Q4 Report"
        ),
    )

# Poll until embedded
while doc.status != "embedded":
    doc = files_retrieve.sync(client=client, id=doc.id)
    if doc.status == "failed":
        raise RuntimeError("ingestion failed")
    time.sleep(2)

# Manage files
docs = files_list.sync(client=client, workspace_id=ws.id)
files_destroy.sync(client=client, id=doc.id)
```

## Async jobs & polling

`parse` and `extract` queue background jobs when passed `options={"async": True}`.
Poll the job with `parse_retrieve` / `extract_retrieve` until it reaches a terminal
state — handy for large documents that would otherwise time out:

```python
import time

from context212_sdk.api.extract import extract, extract_retrieve
from context212_sdk.models import ExtractRequest

job = extract.sync(
    client=client,
    body=ExtractRequest(
        document="https://example.com/big-scan.pdf",
        schema=schema,
        options={"async": True},
    ),
)

while True:
    data = extract_retrieve.sync(client=client, job_id=job.id)
    if data.status == "succeeded":
        print(data.result)
        break
    if data.status == "failed":
        raise RuntimeError(f"extract job {job.id} failed")
    time.sleep(2)
```

## Tags

Tags scope `ask`/`search` to documents carrying them (`tag_id`, OR-matched).

```python
from context212_sdk.api.files import files_tags_create, files_tags_destroy
from context212_sdk.api.search import search
from context212_sdk.api.tags import add_tags
from context212_sdk.models import SearchRequest

tag = add_tags.sync(client=client, body=...)
files_tags_create.sync(client=client, id=doc.id, body=...)

hits = search.sync(
    client=client, body=SearchRequest(query="indemnification", tag_id=[tag.id])
)

files_tags_destroy.sync(client=client, id=doc.id, tag_id=tag.id)
```

## Error handling

Every endpoint function has a `_detailed` variant returning a `Response` with
`status_code`, `headers`, `content`, and `parsed`. The plain `sync`/`asyncio`
variants return just the parsed model (or `None` for undocumented statuses):

```python
resp = search.sync_detailed(client=client, body=SearchRequest(query="..."))
if resp.status_code != 200:
    print("request failed:", resp.content)
```

Set `raise_on_unexpected_status=True` on the client to raise
`errors.UnexpectedStatus` instead.

## Agent frameworks

Context212 drops into any agent stack as a **retrieval tool** — and ships a
ready-made **MCP server** at `https://api.context212.com/mcp` (streamable HTTP), so
MCP-capable agents (Claude Code, Cursor, …) get parse/search/extract tools with zero
code. Connect with your API key and you're done.

For code-first frameworks, wrap a `search` call:

```python
def context212_search(query: str) -> str:
    """Search the company's document corpus for passages relevant to the query."""
    resp = search.sync(
        client=client, body=SearchRequest(query=query, workspace_id=[42], max_results=5)
    )
    return "\n\n".join(f"[{r.source.filename}] {r.content}" for r in resp.results)
```

### LangChain

```python
from langchain_core.tools import tool

context212_tool = tool(context212_search)  # name + description come from the function
# bind it: llm.bind_tools([context212_tool]), or pass to create_react_agent(...)
```

### LangGraph

```python
from langgraph.prebuilt import create_react_agent

agent = create_react_agent(model="openai:gpt-5", tools=[context212_tool])
```

### OpenAI Agents SDK

```python
from agents import Agent, function_tool

agent = Agent(name="Search", tools=[function_tool(context212_search)])
```

## Develop

Regenerate the client from the live OpenAPI schema and rebuild:

```bash
sh scripts/regenerate.sh
```

The `fetch_spec.py` normalization step fixes two FastAPI schema quirks the generator
rejects (invalid `{}` defaults on nullable objects, and binary uploads missing
`format: binary`).
