Context Reference

Last updated: 2026-05-07 06:53:50

sdk context reference

Context Reference¶

The Context object is passed to every command and provides access to platform services, storage, integrations, and request metadata.

Overview¶

from huitzo_sdk import command, Context

@command("example", namespace="demo")
async def example(args: Args, ctx: Context) -> dict:
    # Access storage
    await ctx.storage.save("key", {"data": "value"})

    # Use integrations
    response = await ctx.llm.complete("Summarize this text...")

    # Access metadata
    print(f"User: {ctx.user_id}")
    print(f"Tenant: {ctx.tenant_id}")

    return {"response": response}

Context Properties¶

Identity & Metadata¶

Property	Type	Description
`ctx.user_id`	`UUID`	Current user's ID
`ctx.tenant_id`	`UUID`	Current tenant's ID
`ctx.session_id`	`UUID`	Current session ID
`ctx.correlation_id`	`str`	Request correlation ID for tracing
`ctx.command_name`	`str`	Name of the executing command
`ctx.namespace`	`str`	Namespace of the executing command
`ctx.command_version`	`str`	Version of the executing command

Deployment Mode¶

Property	Type	Description
`ctx.deployment_mode`	`DeploymentMode`	Current deployment mode
`ctx.local_storage`	`Optional[StorageBackend]`	Local storage for edge mode (future)
`ctx.inference_backend`	`Optional[InferenceBackend]`	Local inference for edge mode (future)

The deployment_mode property indicates the current execution environment:

from huitzo_sdk import DeploymentMode

@command("check-mode", namespace="utils")
async def check_mode(args: Args, ctx: Context) -> dict:
    return {
        "mode": ctx.deployment_mode.value,
        "is_cloud": ctx.deployment_mode == DeploymentMode.CLOUD,
        "is_self_hosted": ctx.deployment_mode == DeploymentMode.SELF_HOSTED,
        "is_edge": ctx.deployment_mode == DeploymentMode.EDGE,
    }

Deployment Modes:

Mode	Description	Status
`CLOUD`	Huitzo-managed multi-tenant infrastructure	Production
`SELF_HOSTED`	Customer-managed single-tenant deployment	Production (Year 1)
`EDGE`	Offline-capable edge device	Design phase (Year 3+)

Most commands should work identically across all deployment modes. Use ctx.deployment_mode only when behavior must differ (e.g., offline sync handling).

Example: Accessing Metadata¶

@command("whoami", namespace="utils")
async def whoami(args: Args, ctx: Context) -> dict:
    return {
        "user_id": str(ctx.user_id),
        "tenant_id": str(ctx.tenant_id),
        "session_id": str(ctx.session_id),
        "correlation_id": ctx.correlation_id,
        "command": f"{ctx.namespace}.{ctx.command_name}",
        "version": ctx.command_version,
    }

Services¶

ctx.storage¶

Tenant-isolated key-value storage backed by a pluggable storage backend (PostgreSQL JSONB for cloud/self-hosted, SQLite for edge in future).

# pseudocode — ctx.storage usage

await ctx.storage.save(key, value)              # save data
await ctx.storage.save(key, value, ttl=3600)    # save with expiration (seconds)
data = await ctx.storage.get(key)               # get data (returns None if missing)
data = await ctx.storage.get(key, default={})   # get with default
keys = await ctx.storage.list(prefix="...")      # list keys by prefix
await ctx.storage.delete(key)                    # delete
exists = await ctx.storage.exists(key)           # check existence

Batch Operations:

# pseudocode — batch storage operations

await ctx.storage.save_many({key1: val1, key2: val2, ...})
results = await ctx.storage.get_many([key1, key2, ...])
await ctx.storage.delete_many([key1, key2, ...])

Query by Metadata:

# pseudocode — query by metadata
results = await ctx.storage.query(prefix="...", metadata={...}, limit=100)

See: Storage Reference

ctx.llm¶

AI/LLM completion service.

# pseudocode — ctx.llm usage

# Simple completion
response = await ctx.llm.complete(prompt="...", model="<model-name>")

# With system message
response = await ctx.llm.complete(prompt=..., system="...", model="<model-name>")

# Structured output (returns validated Pydantic model instance)
response = await ctx.llm.complete(prompt=..., response_format="json", schema=YourModel)

# Streaming
async for chunk in ctx.llm.stream(prompt="...", model="<model-name>"):
    # process each chunk
    ...

See: Integrations Reference

ctx.email¶

Email sending service.

# pseudocode — ctx.email usage

# Simple email
await ctx.email.send(to="...", subject="...", body="...")

# HTML email
await ctx.email.send(to="...", subject="...", html="<h1>...</h1>")

# With attachments
await ctx.email.send(to="...", subject="...", body="...",
    attachments=[{"filename": "report.pdf", "content": bytes_data}])

# Multiple recipients
await ctx.email.send(to=["[email protected]", "[email protected]"], cc=["[email protected]"], ...)

ctx.files¶

File reading and writing service.

# Read Excel file
df = await ctx.files.read_excel("uploads/data.xlsx")
df = await ctx.files.read_excel("uploads/data.xlsx", sheet="Sales")

# Read CSV
df = await ctx.files.read_csv("uploads/data.csv")

# Read JSON
data = await ctx.files.read_json("uploads/config.json")

# Write file
await ctx.files.write("output/report.csv", df.to_csv())
await ctx.files.write("output/data.json", json.dumps(data))

# Get file info
info = await ctx.files.info("uploads/data.xlsx")
# Returns: {"size": 1024, "created": datetime, "modified": datetime}

# List files
files = await ctx.files.list("uploads/")

ctx.http¶

HTTP client for external API calls.

# GET request
response = await ctx.http.get("https://api.example.com/data")

# With headers
response = await ctx.http.get(
    "https://api.example.com/data",
    headers={"Authorization": "Bearer token"}
)

# POST request
response = await ctx.http.post(
    "https://api.example.com/submit",
    json={"name": "value"}
)

# With timeout
response = await ctx.http.get(
    "https://slow-api.example.com/data",
    timeout=60
)

# Form data
response = await ctx.http.post(
    "https://api.example.com/upload",
    data={"field": "value"},
    files={"file": file_bytes}
)

Domain Restrictions¶

HTTP requests are restricted to domains declared in your pack's huitzo.yaml:

services:
  http:
    allowed_domains:
      - "api.example.com"        # Exact match
      - "*.trusted-domain.org"   # Wildcard subdomain

Enforcement: - Requests to unlisted domains raise HTTPSecurityError - Wildcards match any subdomain (e.g., *.example.com matches api.example.com) - Protocol is always HTTPS in production

Error Handling:

from huitzo_sdk.errors import HTTPSecurityError

try:
    response = await ctx.http.get("https://blocked-domain.com/data")
except HTTPSecurityError as e:
    ctx.log.error(f"Domain not allowed: {e.domain}")

ctx.cron¶

Cron schedule metadata for commands triggered by the scheduler.

# pseudocode — ctx.cron usage

# Check if this is a scheduled run
if ctx.cron.is_scheduled:
    schedule = ctx.cron.schedule         # e.g., "0 8 * * 1-5"
    scheduled_at = ctx.cron.scheduled_at # datetime when this run was scheduled (UTC)

# Query next/last run times
next_run = await ctx.cron.get_next_run()
last_run = await ctx.cron.get_last_run()

# Query another command's schedule
next_digest = await ctx.cron.get_next_run("@scope/pack/digest")

Properties:

Property	Type	Description
`ctx.cron.is_scheduled`	`bool`	`True` if triggered by scheduler, `False` for on-demand
`ctx.cron.schedule`	`str \\| None`	Cron expression (e.g., `"0 8 * * 1-5"`)
`ctx.cron.scheduled_at`	`datetime \\| None`	When this run was scheduled (UTC)

Methods:

Method	Returns	Description
`await ctx.cron.get_next_run(command_namespace?)`	`datetime \\| None`	Next scheduled run time (UTC)
`await ctx.cron.get_last_run(command_namespace?)`	`datetime \\| None`	Last scheduled run time (UTC)

Requirements: Declare cron in your pack's services section and set a schedule field on the command in huitzo.yaml. See Manifest Reference.

ctx.telegram¶

Telegram messaging service.

# Send message
await ctx.telegram.send(
    chat_id="123456789",
    message="Hello from Huitzo!"
)

# With formatting
await ctx.telegram.send(
    chat_id="123456789",
    message="*Bold* and _italic_",
    parse_mode="Markdown"
)

# Send document
await ctx.telegram.send_document(
    chat_id="123456789",
    document=pdf_bytes,
    filename="report.pdf",
    caption="Your weekly report"
)

ctx.ssh¶

Remote command execution on user-registered SSH targets (GPU clusters, custom servers, on-premise machines).

# Execute a command on a remote server
result = await ctx.ssh.run("gpu-cluster", "nvidia-smi --query-gpu=name,memory.used --format=csv")
print(result.stdout)     # Command output
print(result.exit_code)  # 0 = success

# With custom timeout
result = await ctx.ssh.run("gpu-cluster", "python train.py", timeout=3600)

# Check exit code (non-zero does NOT auto-raise)
if result.exit_code != 0:
    ctx.log.error(f"Command failed: {result.stderr}")

Key Points: - target is the human-readable name the user registered (e.g., "gpu-cluster") - Returns SSHResult with stdout, stderr, exit_code - Output truncated at 1 MB per stream - Pack must declare allowed targets in huitzo.yaml under ssh_targets

Error Handling:

from huitzo_sdk.errors import SSHError

try:
    result = await ctx.ssh.run("gpu-cluster", "nvidia-smi")
except SSHError as e:
    ctx.log.error(f"SSH failed on {e.target}: {e.message}")

See: SSH Integration Reference

ctx.mcp¶

Access to MCP (Model Context Protocol) servers configured in the pack manifest. MCP servers expose external tools that become callable from your commands—no LLM required.

# Call an MCP tool
result = await ctx.mcp.call(
    server="github",
    tool="create_issue",
    arguments={
        "owner": "acme",
        "repo": "project",
        "title": "Bug report"
    }
)

# Attribute access syntax (alternative)
result = await ctx.mcp.github.create_issue(
    owner="acme",
    repo="project",
    title="Bug report"
)

# List available tools from a server
tools = await ctx.mcp.list_tools("github")
for tool in tools:
    print(f"{tool.name}: {tool.description}")

# Get tool schema
schema = await ctx.mcp.get_tool_schema("github", "create_issue")

# Check configured servers
servers = ctx.mcp.servers  # dict[str, MCPServerInfo]

Key Points: - MCP servers are configured in huitzo.yaml under mcp_servers - Tools are discovered automatically on first use - STDIO (subprocess) and HTTP (remote) transports supported - Arguments are validated against the tool's JSON Schema - Connections are lazily initialized and pooled

Error Handling:

from huitzo_sdk.errors import MCPConnectionError, MCPToolError

try:
    result = await ctx.mcp.call("github", "create_issue", {...})
except MCPConnectionError:
    ctx.log.error("GitHub MCP server unavailable")
except MCPToolError as e:
    ctx.log.error(f"Tool failed: {e.message}")

See: MCP Reference for complete API documentation.

ctx.log¶

Structured logging service.

# Log levels
ctx.log.debug("Detailed debug info")
ctx.log.info("Operation completed")
ctx.log.warning("Something might be wrong")
ctx.log.error("An error occurred")

# With context
ctx.log.info("Processing file", extra={
    "filename": args.filename,
    "size": file_size,
    "rows": row_count
})

# Automatic correlation
# All logs automatically include correlation_id, user_id, tenant_id

Environment & Configuration¶

ctx.env (Not yet available)¶

Not yet available in this SDK version. Accessing ctx.env raises ConfigurationError at runtime.

Access environment variables (read-only).

# Get environment variable
api_key = ctx.env.get("MY_API_KEY")

# With default
debug = ctx.env.get("DEBUG", "false")

# Required (raises if missing)
secret = ctx.env.require("REQUIRED_SECRET")

ctx.config (Not yet available)¶

Not yet available in this SDK version. Accessing ctx.config raises ConfigurationError at runtime.

Access pack-specific configuration.

# Get configuration value
setting = ctx.config.get("feature_enabled")

# Nested configuration
db_host = ctx.config.get("database.host")

# With type
timeout = ctx.config.get("timeout", type=int, default=60)

ctx.secrets¶

Access user-provided secrets (API keys, credentials) for external services. User secrets are scoped to user + pack and are distinct from platform environment variables.

# Get secret (returns None if not set)
api_key = ctx.secrets.get("FINANCIAL_API_KEY")

# Require secret (raises SecretsError if not set)
api_key = ctx.secrets.require("FINANCIAL_API_KEY")

# Check existence
if ctx.secrets.exists("PREMIUM_API_KEY"):
    premium_key = ctx.secrets.get("PREMIUM_API_KEY")

Secrets vs Environment Variables¶

Method	Source	Scope	Use Case
`ctx.env.get()`	Platform environment	All packs	Platform API keys (OPENAI_API_KEY)
`ctx.secrets.get()`	User-provided	User + Pack	User's external API keys

Example: Using User Secrets¶

from huitzo_sdk import command, Context
from huitzo_sdk.errors import SecretsError, ExternalAPIError

@command("sync-data", namespace="integrations")
async def sync_data(args: Args, ctx: Context) -> dict:
    """Sync data from user's external service."""

    # Require the user's API key
    try:
        api_key = ctx.secrets.require("EXTERNAL_API_KEY")
    except SecretsError:
        return {
            "error": "Missing API key",
            "help": "Configure your API key in Settings → Pack Secrets"
        }

    # Use the key to call external service
    try:
        response = await ctx.http.get(
            "https://api.external.com/data",
            headers={"Authorization": f"Bearer {api_key}"}
        )
    except HTTPError as e:
        if e.status_code == 401:
            raise ExternalAPIError(
                service="external-api",
                message="Invalid API key. Please verify your key in Settings."
            )
        raise

    return {"synced": len(response["data"])}

Secret Management¶

Users manage their secrets through: - Dashboard: Settings → Pack Secrets → [Pack Name] - CLI: huitzo secrets set PACK_NAME KEY_NAME VALUE

Pack developers declare required and optional secrets in huitzo.yaml:

secrets:
  user_required:
    - name: "EXTERNAL_API_KEY"
      description: "Your API key from external.com"
      help_url: "https://external.com/api-keys"

See Pack Manifest - Secrets and Secrets Management for complete documentation.

Advanced Usage¶

Intra-Pack Command Execution¶

Commands within the same pack can invoke each other using ctx.commands.execute(). This enables composable workflows where higher-level commands build on lower-level ones without duplicating logic.

# pseudocode — intra-pack command composition

@command("compare-visits", namespace="claims", timeout=120)
async def compare_visits(args, ctx):
    # Call get-visit within the same pack
    actual = await ctx.commands.execute(
        "get-visit",
        {"pdv_id": args.pdv_id, "visit_id": args.actual_id},
    )
    anterior = await ctx.commands.execute(
        "get-visit",
        {"pdv_id": args.pdv_id, "visit_id": args.anterior_id},
    )
    # Score using loaded data...
    return {"score": compare(actual, anterior)}

`ctx.commands.execute()` Reference¶

# pseudocode — API signature
result = await ctx.commands.execute(
    command_name,        # short name within this pack (e.g., "get-visit")
    args,                # arguments dict
    timeout=None,        # override timeout (optional)
)

Execution Behavior¶

Concern	Behavior
Scope	Same-pack only — cannot call commands in other packs
Routing	Always inline — never dispatched to Celery
Context	Inherits parent (same user_id, tenant_id, storage scope)
Timeout	Deducted from parent's remaining budget
Circular calls	Detected and rejected (max depth 10)

Error Handling¶

# pseudocode — error handling for intra-pack calls
from huitzo_sdk.errors import (
    CircularCommandError,
    CommandNotFoundError,
    CommandTimeoutError,
    CommandError,
)

try:
    result = await ctx.commands.execute("inner-cmd", args={...})
except CommandNotFoundError:
    # command does not exist in this pack
    ...
except CircularCommandError:
    # A -> B -> A detected
    ...
except CommandTimeoutError:
    # parent's timeout budget exhausted
    ...
except CommandError:
    # inner command raised an error
    ...

Pack-to-Pack Command Execution¶

Packs can execute commands from other installed packs using ctx.execute(). This enables composable workflows where packs build on each other's capabilities.

@command("generate-report", namespace="reporting")
async def generate_report(args: ReportArgs, ctx: Context) -> dict:
    # Call command from analytics pack
    analysis = await ctx.execute(
        pack="analytics",
        command="analyze-data",
        args={"dataset": args.dataset_id, "metrics": ["revenue", "growth"]}
    )

    # Call command from email pack
    await ctx.execute(
        pack="notifications",
        command="send-email",
        args={
            "to": args.recipient,
            "subject": "Your Report",
            "body": format_report(analysis)
        }
    )

    return {"status": "sent", "analysis_id": analysis["id"]}

`ctx.execute()` Reference¶

result = await ctx.execute(
    pack: str,           # Target pack namespace
    command: str,        # Command name
    args: dict,          # Command arguments
    timeout: int = None, # Override timeout (optional)
)

Execution Behavior¶

User Identity: Executes as the original user (preserves user_id, tenant_id)
Correlation: Links to parent via parent_correlation_id for tracing
Permissions: Respects the called command's permission requirements
Billing: Counts toward the original user's usage metrics

Concurrent Job Tracking¶

Cross-pack calls count toward the user's concurrent job limit:

# User has limit of 5 concurrent jobs
# Pack A calls Pack B → counts as 2 jobs total

# In v2: Tracked for observability, not enforced
# Future: May enforce limits at subscription tier level

Error Handling¶

from huitzo_sdk.errors import PackExecutionError

@command("composite-task", namespace="workflows")
async def composite_task(args: Args, ctx: Context) -> dict:
    try:
        result = await ctx.execute("other-pack", "some-command", args={...})
    except PackExecutionError as e:
        ctx.log.error(f"Cross-pack call failed: {e.message}")
        # e.pack, e.command, e.original_error available
        raise

Best Practices¶

Check pack availability - Target pack must be installed for the tenant
Handle failures gracefully - Cross-pack calls can fail independently
Avoid circular calls - Pack A calling Pack B calling Pack A will fail
Consider timeouts - Parent command timeout should account for child execution

Transaction Support¶

@command("atomic-operation", namespace="data")
async def atomic_operation(args: Args, ctx: Context) -> dict:
    async with ctx.storage.transaction():
        # All operations in this block are atomic
        await ctx.storage.save("key1", value1)
        await ctx.storage.save("key2", value2)

        if some_condition:
            raise Exception("Rollback!")  # Both saves are rolled back

    return {"status": "committed"}

Scoped Storage¶

@command("user-data", namespace="profile")
async def user_data(args: Args, ctx: Context) -> dict:
    # Default: scoped to tenant + user + pack
    await ctx.storage.save("preference", value)

    # Tenant-level (shared across users)
    await ctx.storage.save("tenant:setting", value, scope="tenant")

    # Pack-level only (not user-specific)
    await ctx.storage.save("cache:data", value, scope="pack")

Request Context¶

@command("with-headers", namespace="api")
async def with_headers(args: Args, ctx: Context) -> dict:
    # Access original request info
    client_ip = ctx.request.client_ip
    user_agent = ctx.request.headers.get("user-agent")

    return {"ip": client_ip, "ua": user_agent}

Best Practices¶

1. Always Use ctx.log¶

# ✅ Good: Use context logger
ctx.log.info("Processing started", extra={"items": len(items)})

# ❌ Bad: Using print
print(f"Processing {len(items)} items")

2. Handle Missing Data¶

# ✅ Good: Use defaults
data = await ctx.storage.get("key", default={})

# ❌ Bad: Assuming data exists
data = await ctx.storage.get("key")
process(data)  # Might be None!

3. Scope Storage Appropriately¶

# ✅ Good: User-specific data uses default scope
await ctx.storage.save("user:settings", settings)

# ✅ Good: Shared cache uses tenant scope
await ctx.storage.save("cache:exchange-rates", rates, scope="tenant")

Proposed Extensions¶

The following ctx.* extensions are architecturally specified but not yet implemented. See the linked architecture docs for full specifications.

ctx.integrations (Proposed)¶

Pack-developer-created service connectors, accessed by name. See Component Model.

# pseudocode — pack-provided integration access

@command("get-accounts", namespace="sales")
async def get_accounts(args, ctx):
    # Named access (primary pattern)
    accounts = await ctx.integrations.salesforce.query("SELECT Id FROM Account")
    return {"accounts": accounts}

@command("sync-crm", namespace="enterprise")
async def sync_crm(args, ctx):
    # Type-based resolution (advanced — dependency injection)
    crm = ctx.resolve(SalesforceIntegration)
    await crm.sync(args.records)

Integrations are lazy-initialized on first access (configure -> validate -> connect). Built-in integrations (ctx.llm, ctx.email, etc.) remain unchanged.

ctx.config — Hierarchical (Proposed)¶

Configuration with cascading overrides. See Component Model — Hierarchical Configuration.

# pseudocode — hierarchical config access

@command("generate-report", namespace="analytics")
async def generate_report(args, ctx):
    # Resolves through cascade: command -> integration -> pack -> platform
    timeout = ctx.config.get("http.timeout", default=30)

    # Explicit scope access (bypass cascade)
    platform_limit = ctx.config.get("http.timeout", scope="platform")

Cascade order: Platform defaults -> Pack manifest -> Integration config -> Command config -> User/tenant overrides.

ctx.events (Proposed)¶

Async pack-to-pack communication via the system event bus. See Event Bus.

# pseudocode — event bus API

@command("process-email", namespace="scanner")
async def process_email(args, ctx):
    invoice_data = await extract_invoice(args.email)

    # Fire-and-forget — any subscribed pack handles it
    await ctx.events.emit("invoice.received", {
        "sender": args.email.sender,
        "amount": invoice_data["total"],
    })

Events are tenant-scoped with at-least-once delivery. Handlers must be idempotent.

ctx.files — VFS Evolution (Proposed)¶

Mountable file system drivers extending the current ctx.files API. See Virtual File System.

# pseudocode — VFS mount access

@command("merge-reports", namespace="analytics")
async def merge_reports(args, ctx):
    # Reads from S3 via mounted driver
    s3_data = await ctx.files.read_csv("/mnt/s3-reports/q1-sales.csv")

    # Default mount (backward compatible — works exactly as today)
    local_data = await ctx.files.read_csv("data/local.csv")

Mount points are declared in the pack manifest. The current ctx.files API is fully backward compatible.

ctx.pipeline (Proposed)¶

Declarative command piping with streaming. See Piping Protocol.

# pseudocode — programmatic pipeline

@command("run-etl", namespace="data")
async def run_etl(args, ctx):
    pipeline = ctx.pipeline.create("etl-job")
    pipeline.add_stage("source:fetch-data", config={"source": args.source})
    pipeline.add_stage("filter:clean", config={"rules": args.rules})
    pipeline.add_stage("action:analyze", config={"model": "gpt-4o"})

    result = await pipeline.execute()
    return {"pipeline_id": result.id, "rows": result.metadata["total"]}

Pipelines execute as a single correlation unit with streaming backpressure.

Context Reference¶

Overview¶

Context Properties¶

Identity & Metadata¶

Deployment Mode¶

Example: Accessing Metadata¶

Services¶

ctx.storage¶

ctx.llm¶

ctx.email¶

ctx.files¶

ctx.http¶

Domain Restrictions¶

ctx.cron¶

ctx.telegram¶

ctx.ssh¶

ctx.mcp¶

ctx.log¶

Environment & Configuration¶

ctx.env (Not yet available)¶

ctx.config (Not yet available)¶

ctx.secrets¶

Secrets vs Environment Variables¶

Example: Using User Secrets¶

Secret Management¶

Advanced Usage¶

Intra-Pack Command Execution¶

ctx.commands.execute() Reference¶

Execution Behavior¶

Error Handling¶

Pack-to-Pack Command Execution¶

ctx.execute() Reference¶

Execution Behavior¶

Concurrent Job Tracking¶

Error Handling¶

Best Practices¶

Transaction Support¶

Scoped Storage¶

Request Context¶

Best Practices¶

1. Always Use ctx.log¶

2. Handle Missing Data¶

3. Scope Storage Appropriately¶

Proposed Extensions¶

ctx.integrations (Proposed)¶

ctx.config — Hierarchical (Proposed)¶

ctx.events (Proposed)¶

ctx.files — VFS Evolution (Proposed)¶

ctx.pipeline (Proposed)¶

See Also¶

`ctx.commands.execute()` Reference¶

`ctx.execute()` Reference¶