Core SDK Reference

ContextUnit

The ContextUnit is the atomic unit of data exchange across the entire ContextUnity mesh. Instead of passing fragmented strings or dicts between services, every request and response is wrapped in a ContextUnit.

This ensures that metadata (like provenance tracing) travels alongside the payload, enabling full observability.

from contextunity.core import ContextUnit

unit = ContextUnit(
    payload={"key": "value"},
    provenance=["service:action"],
)

# Serialize to protobuf
proto = unit.to_protobuf(pb2_module)

# Deserialize from protobuf
unit = ContextUnit.from_protobuf(proto)

# JSON serialization
json_data = unit.model_dump()

ContextToken & TokenBuilder

The ContextToken provides stateless authentication. Rather than hitting a database to check if a request is authorized, the TokenBuilder cryptographically signs the token.

• Minting: Creates a root token for a tenant with specific permissions.
• Validation: Ensures the token is mathematically valid, not expired, and has the required permission (used extensively by the SecurityGuard).
• Attenuation: Allows a service to create a “lesser” token (delegation) for a background worker, restricting its permissions to only what is strictly necessary.

from contextunity.core.tokens import ContextToken, TokenBuilder

# Create a builder
builder = TokenBuilder(enabled=True)

# Mint a root token
token = builder.mint_root(
    user_ctx={},
    permissions=["brain:read", "brain:write"],
    ttl_s=3600,
    allowed_tenants=["my_project"],
    user_id="user_123",
)

# Validate
token.has_permission("brain:read")    # True
token.can_access_tenant("my_project") # True
token.is_expired()                    # False

# Attenuate (delegate with fewer permissions)
child = builder.attenuate(
    token,
    permissions=["brain:read"],  # Subset only
    agent_id="sub-agent",
)

# Verify (raises PermissionError if invalid)
builder.verify(token, required_permission="brain:read")

SharedConfig

To prevent “hidden” environmental variables spread across the codebase, contextunity.core enforces centralized configuration via Pydantic. By using SharedConfig, all loaded environment variables are strictly typed and validated at startup.

If a required variable is missing or malformed, the application fails fast during boot rather than crashing later in production.

from contextunity.core.config import SharedConfig, load_shared_config_from_env

# Load from environment variables
config = load_shared_config_from_env()

# Access validated settings
print(config.log_level)       # LogLevel.INFO
print(config.redis_url)       # None or "redis://..."
print(config.otel_enabled)    # False
print(config.security.enabled) # False

Note

Never use os.environ.get() directly. Use load_shared_config_from_env() or extend SharedConfig — it provides validation and documentation.

Logging

All ContextUnity services must use structured, unified logging. The get_contextunit_logger ensures logs are formatted correctly for DataDog or ELK stacks.

More importantly, it forces the use of safe_log_value() to prevent accidental leakage of API Keys, LLM Provider tokens, or PII into the observability stack.

from contextunity.core.logging import get_contextunit_logger, safe_log_value

logger = get_contextunit_logger(__name__)

# Safe logging of sensitive values
logger.info("API Key: %s", safe_log_value(api_key))

The safe_log_value() helper automatically masks API keys, tokens, and passwords in log output.

gRPC Utilities

from contextunity.core.grpc_utils import create_channel_sync

# Create a gRPC channel with standard configuration
channel = create_channel_sync("localhost:50051")

# Use with service stubs
from contextunity.core import brain_pb2_grpc
stub = brain_pb2_grpc.BrainServiceStub(channel)

Exception Types

from contextunity.core.exceptions import (
    ContextUnityError,          # Base exception for all ContextUnity errors
    ConfigurationError,         # Invalid configuration
    SecurityError,              # Security/auth failures
    RetrievalError,             # Knowledge retrieval failures
    ProviderError,              # LLM/storage provider failures
    ConnectorError,             # Data source connector failures
    ModelError,                 # LLM model errors
    TransformerError,           # Data transformer failures
    IngestionError,             # Data ingestion failures
    StorageError,               # Storage backend failures
    DatabaseConnectionError,    # Database connection failures
    GraphBuilderError,          # Graph construction failures
    IntentDetectionError,       # Intent classification failures
)