Metadata-Version: 2.4
Name: schift
Version: 0.9.0
Summary: Embedding model migration SDK. Switch models without re-embedding.
Project-URL: Repository, https://github.com/schift-io/schift-py
Project-URL: Bug Tracker, https://github.com/schift-io/schift-py/issues
Author: Platform SDK Maintainers
License-Expression: MIT
Keywords: embeddings,migration,projection,rag,vector-database
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Requires-Python: >=3.9
Requires-Dist: httpx>=0.24
Requires-Dist: numpy>=1.24
Provides-Extra: all
Requires-Dist: chromadb>=0.4; extra == 'all'
Requires-Dist: elasticsearch>=8.0; extra == 'all'
Requires-Dist: openai>=1.40; extra == 'all'
Requires-Dist: pgvector>=0.2; extra == 'all'
Requires-Dist: pinecone-client>=5.0; extra == 'all'
Requires-Dist: psycopg[binary]>=3.1; extra == 'all'
Requires-Dist: pymilvus>=2.4; extra == 'all'
Requires-Dist: pymongo>=4.7; extra == 'all'
Requires-Dist: qdrant-client>=1.7; extra == 'all'
Requires-Dist: redis>=5.0; extra == 'all'
Requires-Dist: weaviate-client>=4.4; extra == 'all'
Provides-Extra: chroma
Requires-Dist: chromadb>=0.4; extra == 'chroma'
Provides-Extra: dev
Requires-Dist: pytest>=7.0; extra == 'dev'
Requires-Dist: ruff>=0.1; extra == 'dev'
Provides-Extra: elasticsearch
Requires-Dist: elasticsearch>=8.0; extra == 'elasticsearch'
Provides-Extra: milvus
Requires-Dist: pymilvus>=2.4; extra == 'milvus'
Provides-Extra: mongodb
Requires-Dist: pymongo>=4.7; extra == 'mongodb'
Provides-Extra: openai
Requires-Dist: openai>=1.40; extra == 'openai'
Provides-Extra: pinecone
Requires-Dist: pinecone-client>=5.0; extra == 'pinecone'
Provides-Extra: postgres
Requires-Dist: pgvector>=0.2; extra == 'postgres'
Requires-Dist: psycopg[binary]>=3.1; extra == 'postgres'
Provides-Extra: qdrant
Requires-Dist: qdrant-client>=1.7; extra == 'qdrant'
Provides-Extra: redis
Requires-Dist: redis>=5.0; extra == 'redis'
Provides-Extra: weaviate
Requires-Dist: weaviate-client>=4.4; extra == 'weaviate'
Description-Content-Type: text/markdown

# Python SDK

Python client and local migration toolkit for workspace search, chat, and embedding migration.

The package currently exposes two public client styles:

- `WorkspaceClient`: modular client for live API operations such as bucket ingest, catalog lookup, embedding, routing, search, usage, and hosted bucket management.
- `Client`: legacy projection client for fitting and downloading `Projection` objects that run locally.

## Install

Base package:

```bash
pip install schift
```

Optional adapters:

```bash
pip install "schift[postgres]"
pip install "schift[qdrant]"
pip install "schift[all]"
```

Development extras:

```bash
pip install -e ".[dev]"
```

## Authentication And Base URLs

The modular client reads `SCHIFT_API_KEY` and a hosted API origin from the
environment. The base URL is a **bare origin** — no `/v1` suffix; version
prefixes belong to request paths:

```bash
export SCHIFT_API_KEY=sch_your_key_here
export SCHIFT_API_URL=https://api.schift.io
```

Equivalent explicit construction:

```python
import os

from schift import WorkspaceClient

client = WorkspaceClient(
    api_key="sch_your_key_here",
    base_url=os.environ["SCHIFT_API_URL"],  # bare origin, e.g. https://api.schift.io
    timeout=60.0,
)
```

Notes:

- `WorkspaceClient` expects a bare origin (e.g. `https://api.schift.io`).
  A legacy trailing `/v1` is tolerated and stripped automatically, so old
  configs keep working — but new configs should not include it.
- If `base_url` is omitted, set `SCHIFT_API_URL` or `SCHIFT_BASE_URL`.
- The legacy `Client` also requires `base_url`, `SCHIFT_API_URL`, or `SCHIFT_BASE_URL`
  (bare origin) and appends `/v1/...` internally.

## Client Lifecycle

`WorkspaceClient` holds a shared `httpx.Client`, so prefer a context manager for short-lived scripts:

```python
from schift import WorkspaceClient

with WorkspaceClient() as client:
    models = client.catalog.list()
```

For long-running processes, keep one `WorkspaceClient` instance and call `close()` during shutdown.

## Bucket Ingest And Search

```python
from schift import WorkspaceClient

with WorkspaceClient() as client:
    bucket = client.buckets.create(name="finance-docs")
    upload = client.buckets.upload(
        bucket["id"],
        [("files", ("q1-report.pdf", open("q1-report.pdf", "rb").read(), "application/pdf"))],
    )

    jobs = client.buckets.list_jobs(bucket_id=bucket["id"])
    resp = client.buckets.search(
        bucket["id"],
        "revenue guidance",
        top_k=5,
        filter={"quarter": "Q1"},
    )

    print(upload["bucket_id"])
    print(jobs[0]["status"] if jobs else "queued")
    print(resp.context)                 # paste-ready context with [n] markers
    print(resp.citations)               # [(1, "report.pdf p.3"), ...]
    print(resp[0].text if resp else "no hits")
```

`search()` returns a typed `SearchResponse` (the answer-ready v2 envelope:
results with citations, warnings, `degraded`) and derives `context` /
`citations` for direct LLM consumption. For raw chunks use the retrieve
surface:

```python
hits = client.buckets.retrieve(bucket["id"], "revenue guidance", top_k=8)
print(hits.status, hits.operational_status)
for hit in hits:
    print(hit.chunk_id, hit.score, hit.text[:80])
```

Use `client.buckets` and `client.query(..., bucket=...)` for retrieval, `POST /v1/chat` for bucket-backed RAG chat with sources, and `POST /v1/chat/completions` for OpenAI-compatible LLM routing without bucket context.

The bucket helper also exposes `list_collections()`, `get_job()`, `list_jobs()`, `wait_for_job()`, and `poll_job()` so scripts can keep ingest orchestration in one place.

## Document Metadata And Connectors

Uploading a document or syncing it in through a connector (Notion, Gmail,
Obsidian, …) attaches metadata you can filter on at search time. This section
covers what gets filled in automatically, which keys are always reserved,
how to add your own metadata, and how to filter with it.

### Reserved metadata keys

These keys are populated structurally (field mapping from the connector's
API response, or Markdown frontmatter parsing) — never inferred by an LLM.
Availability depends on what the source provides; a plain upload with no
frontmatter may leave several of these empty.

| Key | Meaning | Always set? |
|---|---|---|
| `source_kind` | Kind of source (`notion_page`, `email`, `markdown_note`, `upload`, …) | yes |
| `source_created_at` | Original creation time (ISO 8601) | if source provides it |
| `source_modified_at` | Original last-modified time (ISO 8601) | if source provides it |
| `source_author` | Author | if source provides it |
| `source_uri` | Deep link back to the original source | if source provides it |
| `source_filename` | File name | uploads always; connectors if provided |
| `source_tags` | Tags/labels from the source (JSON-encoded array string) | if source provides it |
| `ingested_by` | Who/what ingested it (`upload` or `connector:<provider>`) | yes |
| `ingested_at` | When Schift received the document | yes |
| `cclg_source` | Internal lineage identifier | yes |

Avoid naming your own custom metadata keys the same as these — the system
value takes precedence.

For Markdown uploads (`.md`/`.markdown`), pass `parse_frontmatter=True` to
`upload()` to promote YAML frontmatter (`created`/`date`/`author`/`tags`)
into the reserved keys above:

```python
client.buckets.upload(
    bucket["id"],
    [("files", ("notes.md", open("notes.md", "rb").read(), "text/markdown"))],
    parse_frontmatter=True,
)
```

### Custom metadata

Schift does not read document contents and infer business metadata (e.g.
`department`, `priority`) for you — custom metadata is always explicit:

```python
client.buckets.upload(
    bucket["id"],
    [("files", ("q3-report.pdf", open("q3-report.pdf", "rb").read(), "application/pdf"))],
    metadata={"department": "finance", "priority": "high"},
)
```

Custom metadata is set per document and inherited by every chunk derived
from it. Values must be strings (or JSON-serializable to a string) — the
vector metadata layer stores `dict[str, str]`.

### Filtering by metadata

Reserved keys and your own custom keys are filtered the same way — flat,
top-level keys in `filter=`, not wrapped in a `{"metadata": {...}}` object:

```python
resp = client.buckets.search(
    bucket["id"],
    "revenue guidance",
    filter={
        "source_kind": "email",
        "department": "finance",
        "source_created_at": {"gte": "2026-01-01"},
    },
)
```

See `FilterOperator` above for the full operator set (`eq`/`ne`/`like`/
`prefix`/`in`/`gt`/`gte`/`lt`/`lte`/`exists`). There is no `contains`
operator today — use `like` (`{"like": "%urgent%"}`) for substring matches,
including against JSON-encoded fields like `source_tags`.

## Quickstart

```python
from schift import WorkspaceClient

with WorkspaceClient() as client:
    models = client.catalog.list()
    vector = client.embed(
        "quarterly revenue report",
        model="openai/text-embedding-3-small",
    )

    bucket = client.buckets.create("finance-docs")
    client.db.upsert(
        collection="finance-docs",
        vectors=[
            {
                "id": "doc-1",
                "values": vector.tolist(),
                "metadata": {"source": "q1-report"},
            }
        ],
    )

    hits = client.query(
        "revenue guidance",
        bucket="finance-docs",
        top_k=5,
    )

    print(models[0]["id"] if models else "no models")
    print(hits[0].text if hits else "no hits")  # typed SearchResponse
```

## Module Reference

### `catalog`

List available models and inspect one model by ID.

```python
from schift import WorkspaceClient

with WorkspaceClient() as client:
    models = client.catalog.list()
    model = client.catalog.get("openai/text-embedding-3-small")
```

### `embed`

The embed module is callable for single-text requests and exposes `batch()` for multiple texts.

```python
from schift import WorkspaceClient

with WorkspaceClient() as client:
    one = client.embed(
        "hello world",
        model="openai/text-embedding-3-small",
    )

    many = client.embed.batch(
        texts=["hello", "goodbye"],
        model="openai/text-embedding-3-small",
        dimensions=1024,
    )
```

Return values are NumPy arrays. Both calls go through the OpenAI-shaped
`POST /v1/embeddings` endpoint (`{"input": ..., "model": ...}`).

### `providers`

Register your own LLM provider API key (BYOK) so `client.chat` and `client.completions` call the provider directly instead of consuming the hosted platform shared LLM quota. Supported providers: `"openai"`, `"google"`, `"anthropic"`.

```python
from schift import WorkspaceClient

with WorkspaceClient() as client:
    # Register a key
    client.providers.set(
        "google",
        api_key="AIza...",
        # endpoint_url="https://custom-proxy.example.com",  # optional
    )

    # Check whether a provider is configured (api_key is never returned)
    status = client.providers.get("openai")
    # {"provider": "openai", "configured": True | False, "endpoint_url": None | str}
```

Rotation: a stored BYOK record **shadows** any server-side env var or secret for that provider. To rotate, call `set()` again with the new key — changing env vars alone has no effect on orgs with a BYOK record.

### `routing`

Read or update the server-side routing policy used by Schift when a model is omitted.

```python
from schift import WorkspaceClient

with WorkspaceClient() as client:
    current = client.routing.get()
    updated = client.routing.set(
        primary="openai/text-embedding-3-small",
        fallback="google/gemini-embedding-001",
        mode="failover",
    )
```

### `pii`

Mask Korean PII before sending text to an LLM, vector store, or workflow step.
Use `types` when you want the masking contract to be visible and selectable.

```python
from schift import WorkspaceClient

PII_TYPES = [
    "resident_id",
    "alien_registration",
    "passport",
    "driver_license",
    "address",
    "phone",
    "bank_account",
]

with WorkspaceClient() as client:
    result = client.redact_pii(
        "주민등록번호 850205-1234567, 연락처 010-1234-5678",
        types=PII_TYPES,
    )
    print(result["masked"])

    masked = client.mask(
        "계좌 123-45-678901",
        types=["bank_account"],
    )
```

The default token format returns tokens such as `[PII_PHONE_1]` and a
`reverse_map` so the caller can restore the original values after the workflow
step. Keep that map only in your temporary restore path; Schift does not
persist or gateway-cache it for `pii_type_index` requests. Set
`token_format="label_index"` only when you need legacy tokens such as `[PHONE_1]`.

Send only `masked` into the LLM, agent, vector store, or workflow step. Never
include `reverse_map` in the AI payload; use it only after the AI result returns
if your app needs to restore values.

```python
restored = client.restore_pii(
    "고객 연락처 [PII_PHONE_1]로 안내하세요.",
    result["reverse_map"],
)
```

`restore_pii()` is a stateless API call. It requires an API key and does not
persist or cache the map.

### `bench`

Run a server-side benchmark between two model IDs. The SDK returns `BenchReport`.

```python
from schift import WorkspaceClient

with WorkspaceClient() as client:
    report = client.bench.run(
        source="openai/text-embedding-3-small",
        target="google/gemini-embedding-001",
        data="./eval_queries.jsonl",
    )

    print(report.verdict)
    print(report.summary())
```

### `db`

Manage hosted buckets and write vectors or raw documents.

```python
from schift import WorkspaceClient

with WorkspaceClient() as client:
    vector = client.embed(
        "Schift reduces vector migration downtime.",
        model="openai/text-embedding-3-small",
    )

    bucket = client.db.create_collection(
        name="product-docs",
        dimension=len(vector),
    )

    client.db.upsert(
        collection="product-docs",
        vectors=[
            {
                "id": "doc-1",
                "values": vector.tolist(),
                "metadata": {"title": "Launch plan"},
            }
        ],
    )

    client.db.upsert_text(
        collection="product-docs",
        documents=[
            {
                "id": "doc-2",
                "text": "Schift reduces vector migration downtime.",
                "metadata": {"title": "Overview"},
            }
        ],
        model="openai/text-embedding-3-small",
    )

    stats = client.db.collection_stats("product-docs")
```

Available methods:

- `create_collection(name, dimension)`
- `list_buckets()`
- `get_collection(name)`
- `collection_stats(name)`
- `delete_collection(name)`
- `upsert(collection, vectors)`
- `upsert_text(collection, documents, model)`

### `query`

The query module is callable and supports hosted buckets or an external DB handle.

```python
from schift import WorkspaceClient

with WorkspaceClient() as client:
    hosted = client.query(
        "vector migration rollback plan",
        bucket="product-docs",
        top_k=10,
        rerank=True,
        rerank_top_k=5,
    )

    passthrough = client.query(
        "incident response",
        db="prod-search-db",
        model="openai/text-embedding-3-small",
        top_k=5,
    )
```

### `rerank`

Rerank a list of candidate documents with a cross-encoder style endpoint.

```python
from schift import WorkspaceClient

with WorkspaceClient() as client:
    reranked = client.rerank(
        "incident response",
        documents=[
            {"id": "doc-1", "text": "..."},
            {"id": "doc-2", "text": "..."},
        ],
        top_k=2,
    )
```

### `usage`

Fetch aggregate usage for billing or dashboards.

```python
from schift import WorkspaceClient

with WorkspaceClient() as client:
    usage = client.usage.get(period="30d")
```

The `granularity` parameter is deprecated (the server ignores it) and is no
longer sent. See `docs/USAGE_CONTRACT.md` for the returned keys.

## Workflows

Build and run RAG pipelines as composable DAGs.

### Quick Start

```python
from schift import WorkspaceClient

with WorkspaceClient() as client:
    # Create from template
    wf = client.workflow.create_rag("Product Search")

    # Run with inputs
    result = client.workflow.run(wf.id, inputs={"query": "best laptop"})
    print(result.outputs["answer"])
```

### CRUD

```python
wf = client.workflow.create("My Pipeline", description="Custom RAG")
workflows = client.workflow.list()
wf = client.workflow.get(wf.id)
wf = client.workflow.update(wf.id, name="Renamed")
client.workflow.delete(wf.id)
```

### Building a Graph

```python
wf = client.workflow.create("Custom RAG")

# Add blocks
start = client.workflow.add_block(wf.id, "start", title="Start")
retriever = client.workflow.add_block(wf.id, "retriever", config={
    "bucket": "my-docs",
    "top_k": 5,
    "rerank": True,
    "rerank_top_k": 3,
})
prompt = client.workflow.add_block(wf.id, "prompt_template", config={
    "template": "Answer based on:\n{{results}}\n\nQuestion: {{query}}",
})
llm = client.workflow.add_block(wf.id, "llm", config={
    "model": "openai/gpt-4o-mini",  # or "anthropic/claude-sonnet-4-20250514", "gemini-2.5-flash"
    "temperature": 0.7,
})
end = client.workflow.add_block(wf.id, "end")

# Connect blocks
client.workflow.add_edge(wf.id, start["id"], retriever["id"])
client.workflow.add_edge(wf.id, retriever["id"], prompt["id"])
client.workflow.add_edge(wf.id, prompt["id"], llm["id"])
client.workflow.add_edge(wf.id, llm["id"], end["id"])
```

### Multiple Inputs

The start node forwards all input variables to downstream blocks:

```python
result = client.workflow.run(wf.id, inputs={
    "query": "maternity leave policy",
    "user_id": "u-123",
    "language": "ko",
})
```

Reference variables in prompt templates with `{{variable}}` syntax.

### Validation & Run History

```python
# Validate graph (cycles, missing connections, etc.)
v = client.workflow.validate(wf.id)
print(v.valid, v.errors)

# List past runs
runs = client.workflow.runs(wf.id)
for r in runs:
    print(r.run_id, r.status, r.outputs)
```

### YAML Import / Export

```python
# Export to YAML
yaml_str = client.workflow.to_yaml(wf.id, path="pipeline.yaml")

# Load from file
definition = client.workflow.from_yaml("pipeline.yaml")

# Push YAML to create on server
wf = client.workflow.push_yaml("pipeline.yaml")
```

### Templates

| Method | Template |
|--------|----------|
| `create_rag(name)` | Retriever -> Prompt -> LLM |
| `create_doc_qa(name)` | Document QA with sources |
| `create_chat(name)` | Conversational RAG |
| `create_ocr_ingest(name)` | OCR -> Chunk -> Embed |

### Workflow Runs

```python
from schift import WorkspaceClient

with WorkspaceClient() as client:
    wf = client.workflow.create_rag("Workspace RAG")
    result = client.workflow.run(wf.id, inputs={"query": "hello"})
```

### Hosted Workflow v2 Runs

Use `workflows_v2` for stored and published Workflow v2 definitions. This calls
`POST /v2/workflows/{workflow_id}/run` and keeps `simulate`/`live`, approvals,
and mock binding payloads aligned with the hosted API.

```python
from schift import WorkspaceClient

with WorkspaceClient() as client:
    run = client.workflows_v2.run(
        "wf_v2_123",
        inputs={"invoice": "T-001"},
        mode="live",
        approvals={"review_issue": True},
        mock_bindings={"tax_provider": {"status": "mocked"}},
    )
    print(run["status"])
```

### Block Types

| Category | Types |
|----------|-------|
| Control | `start`, `end`, `conditional`, `loop` |
| Retrieval | `retriever` (with rerank toggle), `reranker` |
| LLM | `llm` (OpenAI/Anthropic/Google), `prompt_template`, `answer` |
| Data | `document_loader`, `chunker`, `embedder`, `text_processor` |
| Integration | `api_call`, `webhook`, `code_executor` |
| Storage | `vector_store`, `cache` |

### API Reference

| Method | Description |
|--------|-------------|
| `workflow.create(name)` | Create workflow |
| `workflow.get(id)` | Get workflow |
| `workflow.list()` | List workflows |
| `workflow.update(id, **kw)` | Update workflow |
| `workflow.delete(id)` | Delete workflow |
| `workflow.run(id, inputs)` | Run workflow |
| `workflow.runs(id)` | List past runs |
| `workflow.validate(id)` | Validate graph |
| `workflow.add_block(id, type, config)` | Add block |
| `workflow.add_edge(id, src, tgt)` | Add edge |
| `workflow.to_yaml(id)` | Export as YAML |
| `workflow.from_yaml(path)` | Load YAML definition |
| `workflow.push_yaml(path)` | Import YAML to server |

## Projection Workflow

Projection creation still lives in the legacy `Client` API. It returns a local `Projection` object that can transform vectors without further API calls.

```python
from schift import Client

legacy = Client(api_key="sch_your_key_here")

# Embed the same sample texts with both providers before fitting.
source_pairs = old_model_embeddings
target_pairs = new_model_embeddings

projection = legacy.fit(
    source=source_pairs,
    target=target_pairs,
    source_model="openai/text-embedding-3-small",
    target_model="google/gemini-embedding-001",
    project_name="openai-to-gemini",
)

converted = projection.transform(source_pairs[:10])
projection.save("./projection-openai-to-gemini")
```

You can later reload a saved projection:

```python
from schift import Projection

projection = Projection.load("./projection-openai-to-gemini")
```

The legacy client also supports:

- `fit(...)`
- `bench(...)`
- `list_projections()`
- `get_projection(project_id)`

## Local Migration With Adapters

The local migration engine reads vectors from a source adapter, applies a `Projection`, and writes to a sink adapter. This path does not depend on hosted collection APIs.

```python
from schift import Projection
from schift.migrate import migrate
from schift.adapters.file import NpyAdapter

projection = Projection.load("./projection-openai-to-gemini")

source = NpyAdapter("old_embeddings.npy")
sink = NpyAdapter("new_embeddings.npy")

result = migrate(
    source=source,
    sink=sink,
    projection=projection,
    batch_size=2048,
    dry_run=True,
)

print(result)
```

The same engine is exposed through `WorkspaceClient().migrate.run(...)`.

### Built-in Adapters

#### NumPy files

```python
from schift.adapters.file import NpyAdapter

adapter = NpyAdapter("embeddings.npy")
```

#### pgvector

Requires `pip install "schift[postgres]"`.

```python
from schift.adapters.pgvector import PgVectorAdapter

adapter = PgVectorAdapter(
    conninfo="postgresql://user:password@localhost/mydb",
    table="documents",
    embedding_column="embedding",
    id_column="id",
)
```

#### Qdrant

Requires `pip install "schift[qdrant]"`.

```python
from schift.adapters.qdrant import QdrantAdapter

adapter = QdrantAdapter(
    url="http://localhost:6333",
    collection="documents",
    api_key=None,
)
```

#### Registry Helpers

```python
from schift.adapters import get_adapter, list_adapters

print(list_adapters())

adapter = get_adapter(
    {
        "type": "npy",
        "path": "embeddings.npy",
    }
)
```

## Errors

Both client styles raise exceptions from `schift.client`:

- `AuthError`
- `QuotaError`
- `SDKError`

Example:

```python
from schift import WorkspaceClient
from schift import AuthError, QuotaError, SDKError

try:
    with WorkspaceClient() as client:
        client.catalog.list()
except AuthError:
    ...
except QuotaError:
    ...
except SDKError:
    ...
```

## Development

From `clients/sdk/python/`:

```bash
python -m pip install -e ".[dev]"
python -m ruff check .
python -m pytest
```

## Source Layout

```text
clients/sdk/python/
├── schift/
│   ├── schift_client.py   # modular SDK entry point
│   ├── client.py          # legacy projection client
│   ├── projection.py      # local projection object
│   ├── adapters/          # npy, pgvector, qdrant
│   ├── workflow.py         # workflow CRUD, blocks, edges, YAML, templates
│   └── *.py               # catalog, embed, routing, db, query, rerank, usage
└── docs/                  # request/response contracts and schemas
```
