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

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

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.

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)

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)

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)

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:

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/

Guides (how-to): docs/guides/

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

FAQ

Troubleshooting

# 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)

Contact: sales@corpusos.com · partners@corpusos.com

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