Core SDK Reference

ContextUnit

The ContextUnit is the atomic data envelope for all gRPC communication. Every request and response across the mesh is wrapped in a ContextUnit that carries payload, provenance tracing, and security scopes.

from contextunity.core import ContextUnit, contextunit_pb2

unit = ContextUnit(
    payload={"key": "value"},
    provenance=["service:action"],
)
proto = unit.to_protobuf(contextunit_pb2)     # → gRPC
unit = ContextUnit.from_protobuf(proto)        # ← gRPC

For the full protocol definition, usage rules, and conformance requirements, see ContextUnit Protocol.

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 get_core_config

# Load from environment variables + YAML config
config = get_core_config()

# Access validated settings
print(config.log_level)       # LogLevel.INFO
print(config.redis.url)       # "redis://localhost:6379/0"
print(config.redis.enabled)   # True

Note

Never use os.environ.get() directly. Use get_core_config() or extend SharedConfig — it provides validation, YAML overlay, and type safety.

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

All exceptions inherit from ContextUnityError and carry stable error codes mapped to gRPC status codes.

from contextunity.core.exceptions import (
    ContextUnityError,          # Base
    ConfigurationError,         # Invalid configuration
    SecurityError,              # Token/auth failures
    PlatformServiceError,       # Cross-service call failures
    LLMExecutionError,          # LLM invocation failures
)
from contextunity.core.grpc_errors import grpc_error_handler  # gRPC decorator

For the complete hierarchy, error code reference, and gRPC mapping table, see Error Handling.

Bootstrap

Consumer projects register with ContextRouter using a single function call:

from contextunity.core.sdk.bootstrap import bootstrap_django

# Django: call from AppConfig.ready()
bootstrap_django(prompts={"planner": PLANNER_PROMPT})

For the full bootstrap pipeline, manifest discovery, and prompt map formats, see Project Bootstrap.