Corpusos

Install the `corpus_sdk` Python package from PyPI. It has no heavy runtime dependencies and supports Python 3.10+.

pip install corpus_sdk

Choose the protocol domain you need (LLM, Embedding, Vector, or Graph) and subclass the corresponding base adapter. Implement the required `_do_*` methods to wrap your provider's SDK.

from corpus_sdk.llm.llm_base import BaseLLMAdapter, LLMCompletion

class MyLLMAdapter(BaseLLMAdapter):
    async def _do_complete(self, messages, model, **kwargs) -> LLMCompletion:
        # wrap your provider SDK here
        ...

Create an `OperationContext` to carry per-request metadata including request ID, tenant isolation, deadline, and cache TTL. Pass it to every adapter call.

from corpus_sdk.llm.llm_base import OperationContext

ctx = OperationContext(
    request_id="req-001",
    tenant="my-team",
    cache_ttl_s=300,
    deadline_ms=30000
)

Use the adapter's async methods (`complete`, `embed`, `upsert`, `query`) directly. The protocol layer handles retries, circuit breaking, and normalized error responses.

result = await adapter.complete(
    messages=[{"role": "user", "content": "Explain vector databases"}],
    model="gpt-4-turbo",
    ctx=ctx
)
print(result.text, result.usage.total_tokens)

Validate your adapter implementation against the full 3,330+ conformance test suite. This ensures your adapter handles all edge cases, error conditions, and protocol requirements correctly.

# Run all conformance tests
make test-all-conformance

# Or test a single protocol
make test-llm-conformance
make test-vector-conformance

Connect the Corpus OS MCP server to your AI agent client to expose standardized LLM and vector capabilities as MCP tools.