corpus-sdk 1.0.0

Protocol-first, vendor-neutral SDK for LLM, Embedding, Vector, and Graph backends.

Corpus OS Protocol and SDK

Reference implementation of the Corpus OS Protocol and SDK — a wire-first, vendor-neutral SDK for interoperable AI frameworks and data backends across four domains: LLM, Embedding, Vector, and Graph.

┌──────────────────────────────────────────────────────────────────────┐
│  Your App / Agents / RAG Pipelines                                   │
│  (LangChain · LlamaIndex · Semantic Kernel · CrewAI · AutoGen · MCP) │
├──────────────────────────────────────────────────────────────────────┤
│  Corpus OS Protocol and SDK.                                         │
│  One protocol · One error taxonomy · One metrics model               │
├──────────┬──────────────┬────────────┬───────────────────────────────┤
│ LLM/v1   │ Embedding/v1 │ Vector/v1  │ Graph/v1                      │
├──────────┴──────────────┴────────────┴───────────────────────────────┤
│  Any Provider: OpenAI · Anthropic · Pinecone · Neo4j · ...           │
└──────────────────────────────────────────────────────────────────────┘

Keep your frameworks. Standardize your infrastructure.

Open-Core Model: The Corpus OS Protocol Suite and SDK are fully open source (Apache-2.0). Corpus Router and official production adapters are commercial, optional, and built on the same public protocols. Using this SDK does not lock you into CORPUS Router.

---

Table of Contents

1. Why CORPUS
2. How CORPUS Compares
3. When Not to Use CORPUS
4. Install
5. Quick Start
6. Domain Examples
7. Core Concepts
8. Error Taxonomy & Observability
9. Performance & Configuration
10. Testing & Conformance
11. Documentation Layout
12. FAQ
13. Contributing
14. License & Commercial Options

---

Why Corpus OS

Modern AI platforms juggle multiple LLM, embedding, vector, and graph backends. Each vendor ships unique APIs, error schemes, rate limits, and capabilities — making cross-provider integration brittle and costly.

The problem:

• Provider proliferation — Dozens of incompatible APIs across AI infrastructure
• Duplicate integration — Different error handling, observability, and resilience patterns rewritten per provider and framework
• Vendor lock-in — Applications tightly coupled to specific backend choices
• Operational complexity — Inconsistent monitoring and debugging across services

Corpus OS provides:

• Stable, runtime-checkable protocols across all four domains
• Normalized errors with retry hints and machine-actionable scopes
• SIEM-safe metrics (low-cardinality, tenant-hashed, no PII)
• Deadline propagation for cancellation and cost control
• Two modes — compose under your own router (thin) or use lightweight built-in infra (standalone)
• Wire-first design — canonical JSON envelopes implementable in any language, with this SDK as reference

Corpus OS is not a replacement for LangChain, LlamaIndex, Semantic Kernel, CrewAI, AutoGen, or MCP. Use those for agent-specific orchestration, agents, tools, and RAG pipelines. Use Corpus OS to standardize the infrastructure layer underneath them. Your app teams keep their frameworks. Your platform team gets one protocol, one error taxonomy, and one observability model across everything.

---

How Corpus OS Compares

Aspect	LangChain / LlamaIndex	OpenRouter	MCP	Corpus OS
Scope	Application framework	LLM unification	Tools & data sources	AI infrastructure protocols
Domains	LLM + Tools	LLM only	Tools + Data	LLM + Vector + Graph + Embedding
Error Standardization	Partial	Limited	N/A	Comprehensive taxonomy
Multi-Provider Routing	Basic	Managed service	N/A	Protocol for any router
Observability	Basic	Limited	N/A	Built-in metrics + tracing
Vendor Neutrality	High	Service-dependent	High	Protocol-first, no lock-in

Who is this for?

• App developers — Keep using your framework of choice. Talk to all backends through Corpus OS protocols. Swap providers or frameworks without rewriting integration code.
• Framework maintainers — Implement one CORPUS adapter per protocol. Instantly support any conformant backend.
• Backend vendors — Implement llm/v1, embedding/v1, vector/v1, or graph/v1 once, run the conformance suite, and your service works with every framework.
• Platform / infra teams — Unified observability: normalized error codes, deadlines, and metrics. One set of dashboards and SLOs across all AI traffic.
• MCP users — The Corpus OS MCP server exposes protocols as standard MCP tools. Any MCP client can call into your infra with consistent behavior.

Integration Patterns

Pattern	How It Works	What You Get
Framework → Corpus OS → Providers	Framework uses Corpus OS as client	Unified errors/metrics across providers
Corpus OS → Framework-as-adapter → Providers	Framework wrapped as Corpus OS adapter	Reuse existing chains/indices as "providers"
Mixed	Both of the above	Gradual migration, no big-bang rewrites

Large teams typically run all three patterns at once.

---

When Not to Use CORPUS

You probably don't need Corpus OS if:

• Single-provider and happy — One backend, fine with their SDK and breaking changes.
• No governance pressure — No per-tenant isolation, budgets, audit trails, or data residency.
• No cross-domain orchestration — Not coordinating LLM + Vector + Graph + Embedding together.
• Quick throwaway prototype — Lock-in, metrics, and resilience aren't worth thinking about yet.

If any of these stop being true, corpus_sdk is the incremental next step.

---

Install

pip install corpus_sdk

Python ≥ 3.10 recommended. No heavy runtime dependencies.

---

⚡ Quick Start

from corpus_sdk.llm.llm_base import BaseLLMAdapter, OperationContext

class QuickAdapter(BaseLLMAdapter):
    async def _do_complete(self, messages, **kwargs):
        return {"text": "Hello from CORPUS!", "model": "quick-demo"}

adapter = QuickAdapter()
ctx = OperationContext(request_id="test-123")
result = await adapter.complete(
    messages=[{"role": "user", "content": "Hi"}], ctx=ctx
)
print(result.text)  # "Hello from CORPUS!"

A complete quick start with all four protocols is in docs/guides/QUICK_START.md.

---

Domain Examples

Minimal viable adapter: Implement _do_capabilities, your core operation (_do_embed, _do_complete, _do_query, etc.), and _do_health. All other methods have safe no-op defaults — you only override what you need.

In all examples, swap Example*Adapter with your actual adapter class that inherits the corresponding base and implements _do_* hooks.

Embeddings

from corpus_sdk.embedding.embedding_base import (
    BaseEmbeddingAdapter, EmbedSpec, OperationContext,
    EmbeddingVector, EmbeddingCapabilities, EmbedResult
)

class ExampleEmbeddingAdapter(BaseEmbeddingAdapter):
    async def _do_capabilities(self) -> EmbeddingCapabilities:
        return EmbeddingCapabilities(
            server="example-embeddings", version="1.0.0",
            supported_models=("example-embed-001",),
            max_batch_size=128, max_text_length=8192,
            supports_normalization=True, normalizes_at_source=False,
            supports_deadline=True, supports_token_counting=False,
        )

    async def _do_embed(self, spec: EmbedSpec, *, ctx=None) -> EmbedResult:
        vec = [0.1, 0.2, 0.3]
        return EmbedResult(
            embedding=EmbeddingVector(vector=vec, text=spec.text,
                                      model=spec.model, dimensions=len(vec)),
            model=spec.model, text=spec.text,
            tokens_used=None, truncated=False,
        )

    async def _do_health(self, *, ctx=None) -> dict:
        return {"ok": True, "server": "example-embeddings", "version": "1.0.0"}

# Usage
async with ExampleEmbeddingAdapter() as adapter:
    ctx = OperationContext(request_id="req-1", tenant="acme")
    res = await adapter.embed(
        EmbedSpec(text="hello world", model="example-embed-001"), ctx=ctx
    )
    print(res.embedding.vector)

LLM

from corpus_sdk.llm.llm_base import (
    BaseLLMAdapter, OperationContext, LLMCompletion,
    TokenUsage, LLMCapabilities
)

class ExampleLLMAdapter(BaseLLMAdapter):
    async def _do_capabilities(self) -> LLMCapabilities:
        return LLMCapabilities(
            server="example-llm", version="1.0.0",
            model_family="gpt-4", max_context_length=8192,
            supports_streaming=True, supports_roles=True,
            supports_json_output=False, supports_parallel_tool_calls=False,
            idempotent_writes=False, supports_multi_tenant=True,
            supports_system_message=True,
        )

    async def _do_complete(self, messages, model, **kwargs) -> LLMCompletion:
        return LLMCompletion(
            text="Hello from example-llm!", model=model,
            model_family="gpt-4",
            usage=TokenUsage(prompt_tokens=5, completion_tokens=5, total_tokens=10),
            finish_reason="stop",
        )

    async def _do_health(self, *, ctx=None) -> dict:
        return {"ok": True, "server": "example-llm", "version": "1.0.0"}

# Usage
async with ExampleLLMAdapter() as adapter:
    ctx = OperationContext(request_id="req-2", tenant="acme")
    resp = await adapter.complete(
        messages=[{"role": "user", "content": "Say hi"}],
        model="example-llm-001", ctx=ctx,
    )
    print(resp.text)

Vector

from corpus_sdk.vector.vector_base import (
    BaseVectorAdapter, VectorCapabilities, QuerySpec,
    QueryResult, Vector, VectorMatch, OperationContext, VectorID
)

class ExampleVectorAdapter(BaseVectorAdapter):
    async def _do_capabilities(self) -> VectorCapabilities:
        return VectorCapabilities(
            server="example-vector", version="1.0.0", max_dimensions=3
        )

    async def _do_query(self, spec: QuerySpec, *, ctx=None) -> QueryResult:
        v = Vector(id=VectorID("v1"), vector=[0.1, 0.2, 0.3],
                   metadata={"label": "demo"}, namespace=spec.namespace)
        return QueryResult(
            matches=[VectorMatch(vector=v, score=0.99, distance=0.01)],
            query_vector=spec.vector, namespace=spec.namespace, total_matches=1,
        )

    async def _do_health(self, *, ctx=None) -> dict:
        return {"ok": True, "server": "example-vector", "version": "1.0.0"}

# Usage
adapter = ExampleVectorAdapter()
ctx = OperationContext(request_id="req-3", tenant="acme")
result = await adapter.query(QuerySpec(vector=[0.1, 0.2, 0.3], top_k=1), ctx=ctx)
print(result.matches[0].score)

Graph

from corpus_sdk.graph.graph_base import (
    BaseGraphAdapter, GraphCapabilities, UpsertNodesSpec,
    Node, GraphID, OperationContext, GraphQuerySpec, GraphQueryResult
)

class ExampleGraphAdapter(BaseGraphAdapter):
    async def _do_capabilities(self) -> GraphCapabilities:
        return GraphCapabilities(
            server="example-graph", version="1.0.0",
            supported_query_dialects=("cypher",),
            supports_stream_query=True, supports_bulk_vertices=True,
            supports_batch=True, supports_schema=True,
        )

    async def _do_query(self, spec: GraphQuerySpec, *, ctx=None) -> GraphQueryResult:
        return GraphQueryResult(
            records=[{"id": 1, "name": "Ada"}],
            summary={"rows": 1}, dialect=spec.dialect,
            namespace=spec.namespace,
        )

    async def _do_health(self, *, ctx=None) -> dict:
        return {"ok": True, "server": "example-graph", "version": "1.0.0"}

# Usage
async with ExampleGraphAdapter() as adapter:
    ctx = OperationContext(request_id="req-4", tenant="acme")
    result = await adapter.upsert_nodes(
        UpsertNodesSpec(nodes=[
            Node(id=GraphID("user:1"), labels=("User",),
                 properties={"name": "Ada"})
        ]),
        ctx=ctx,
    )
    print(f"Upserted {result.upserted_count} nodes")

Full implementations with batch operations, streaming, and multi-cloud scenarios are in docs/guides/ADAPTER_RECIPES.md.

---

Core Concepts

• Protocol vs Base — Protocols define required behavior. Bases implement validation, deadlines, observability, and error normalization. You implement _do_* hooks.
• OperationContext — Carries request_id, idempotency_key, deadline_ms, traceparent, tenant, and cache hints across all operations.
• Wire Protocol — Canonical envelopes (op, ctx, args) and response shapes (ok, code, result) defined in docs/spec/PROTOCOL.md.
• Corpus OS-Compatible — Implementations that honor the envelopes, reserved op strings, and error taxonomy. Validated by the conformance suite.

---

Error Taxonomy & Observability

All domains share a normalized error taxonomy: BadRequest, AuthError, ResourceExhausted, TransientNetwork, Unavailable, NotSupported, DeadlineExceeded, plus domain-specific errors like TextTooLong, ModelOverloaded, DimensionMismatch, and IndexNotReady.

Errors carry machine-actionable hints (retry_after_ms, throttle_scope) so routers and control planes can react consistently across providers. A pluggable MetricsSink protocol lets you bring your own metrics backend. Bases emit one observe per operation, hash tenants before recording, and never log prompts, vectors, or raw tenant IDs.

Full details in docs/spec/ERRORS.md and docs/spec/METRICS.md.

---

Performance & Configuration

Base overhead is typically <10 ms relative to vendor SDK calls: validation <1 ms, metrics <0.1 ms, cache lookup (standalone) <0.5 ms. Async-first design avoids blocking and supports high concurrency. Batch operations (embed_batch, vector upserts, graph batch) are preferred for throughput.

Benchmarks and deployment patterns in docs/guides/IMPLEMENTATION.md.

Modes: thin vs standalone

Once you're ready for production, choose a mode:

Mode	Infra Hooks	When to Use
thin (default)	All no-ops	You have an external control plane (router, scheduler, limiter)
standalone	Deadline enforcement, circuit breaker, token-bucket limiter, in-memory TTL cache	Lightweight deployments without external infra

Use thin under a router to prevent double-stacking resiliency. Use standalone for prototyping or single-service deployments.

---

Testing & Conformance

One-Command Testing

# All protocols at once
make test-all-conformance

# Specific protocols
make test-llm-conformance
make test-vector-conformance
make test-graph-conformance
make test-embedding-conformance

CLI

corpus-sdk verify                        # All protocols
corpus-sdk verify -p llm -p vector       # Selected protocols
corpus-sdk test-llm-conformance          # Single protocol

Direct pytest

pytest tests/ -v --cov=corpus_sdk --cov-report=html

Requirements for "CORPUS-Compatible" certification are in docs/conformance/CERTIFICATION.md.

---

📚 Documentation Layout

Spec (normative): docs/spec/

File	Contents
SPECIFICATION.md	Full protocol suite specification (all domains, cross-cutting behavior)
PROTOCOL.md	Wire-level envelopes, streaming semantics, canonical op registry
ERRORS.md	Canonical error taxonomy & mapping rules
METRICS.md	Metrics schema & SIEM-safe observability
SCHEMA.md	JSON/type shapes
VERSIONING.md	Semantic versioning & compatibility rules

Guides (how-to): docs/guides/

File	Contents
QUICK_START.md	End-to-end flows for all four protocols
IMPLEMENTATION.md	How to implement adapters
ADAPTER_RECIPES.md	Real-world multi-cloud and RAG scenarios
CONFORMANCE_GUIDE.md	How to run & interpret conformance suites

Conformance (testing): docs/conformance/ — Per-protocol test specs, schema conformance, behavioral conformance, and certification requirements.

---

FAQ

Is the SDK open source?

Yes. The SDK (protocols, bases, example adapters) is open source under Apache-2.0. CORPUS Router and official production adapters are commercial.

Do I have to use CORPUS Router?

No. The SDK composes with any router or control plane. CORPUS Router is optional and adheres to the same public protocols.

How does CORPUS relate to LangChain / LlamaIndex / MCP / OpenRouter?

They're complementary. LangChain/LlamaIndex are application frameworks. MCP standardizes tools and data sources. OpenRouter unifies LLM providers. CORPUS standardizes the infrastructure layer (LLM + Vector + Graph + Embedding) underneath all of them with consistent errors, metrics, and capabilities discovery.

Why async-only?

Modern AI workloads require high concurrency. Async-first prevents blocking the event loop. Sync wrappers can be built on top if needed.

What happens if my adapter raises a non-normalized error?

Bases catch unexpected exceptions and record them as UnhandledException in metrics. Wrap provider errors in normalized exceptions for proper handling.

Can CORPUS Router run on-prem?

Yes. Available as managed cloud or on-prem deployment for regulated and air-gapped environments.

---

Troubleshooting

Problem	Solution
Double-stacked resiliency (timeouts firing twice)	Ensure mode="thin" under your router
Circuit breaker opens frequently	Reduce concurrency or switch to thin with external circuit breaker
Cache returns stale results	Verify sampling params in cache key; check cache_ttl_s
DeadlineExceeded on fast operations	Ensure deadline_ms is absolute epoch time, not relative. Check NTP sync.
Health check failures	Inspect _do_health implementation; verify backend reachability and credentials

# Debug mode
import logging
logging.basicConfig(level=logging.DEBUG)
logging.getLogger("corpus_sdk").setLevel(logging.DEBUG)

---

Contributing

git clone https://github.com/corpus/corpus-sdk.git
cd corpus-sdk
pip install -e ".[dev]"
pytest

Follow PEP-8 (ruff/black). Type hints required on all public APIs. Include tests for new features. Maintain low-cardinality metrics — never add PII to extra fields. Observe SemVer.

We especially welcome community adapter contributions for new LLM, vector, graph, and embedding backends.

Community questions: GitHub Discussions preferred.

---

License & Commercial Options

License: Apache-2.0 (LICENSE)

Need	Solution	Cost
Learning / prototyping	corpus_sdk + example adapters	Free (OSS)
Production with your own infra	corpus_sdk + your adapters	Free (OSS)
Production with official adapters	corpus_sdk + Official Adapters	Commercial
Enterprise multi-provider	corpus_sdk + CORPUS Router (Managed or On-Prem)	Commercial

Contact: [e-mail] · [e-mail]

---

Built by the Corpus OS team — wire-level AI infrastructure you integrate once and stop thinking about.