SDK Overview

sdk python overview

Huitzo SDK Overview

The Huitzo SDK (huitzo-sdk) provides everything you need to build Intelligence Packs. It handles the boring stuff—validation, retries, storage, multi-tenancy—so you can focus on business logic.

Installation

pip install huitzo-sdk

Or with uv:

uv add huitzo-sdk

Design Philosophy

The SDK follows a progressive disclosure model:

  1. Simple things are simple – A basic command is ~10 lines of code
  2. Complex things are possible – Full control when you need it
  3. Magic is optional – All automatic behaviors can be configured

What the SDK Does for You

Feature Automatic Configurable
Input validation ✅ From Pydantic models ✅ Custom validators
Retries ✅ 3 retries, exponential backoff ✅ Custom retry policy
Timeouts ✅ 60s default ✅ Per-command timeout
Error handling ✅ Structured errors with debug IDs ✅ Custom error types
Output formatting ✅ Auto-detect JSON/text ✅ Explicit format hints
Logging ✅ Structured JSON logs ✅ Log levels, correlation IDs

What You Control

Feature You Decide
Permission checks Explicit in your code
Data validation logic Your Pydantic models
Business logic 100% yours
Error messages Your custom messages

Quick Start

1. Create a Command

from huitzo_sdk import command, Context
from pydantic import BaseModel

class AddArgs(BaseModel):
    a: int
    b: int

@command("add", namespace="math")
async def add(args: AddArgs, ctx: Context) -> int:
    """Add two numbers together."""
    return args.a + args.b

Intelligent Command (Input → LLM → Structured Output):

from pydantic import BaseModel
from huitzo_sdk import command, Context

class SummarizeArgs(BaseModel):
    text: str
    max_bullets: int = 3

class Summary(BaseModel):
    bullets: list[str]
    sentiment: str
    word_count: int

@command("summarize", namespace="content")
async def summarize(args: SummarizeArgs, ctx: Context) -> Summary:
    """Summarize text into bullet points with sentiment analysis."""
    response = await ctx.llm.complete(
        prompt=f"Summarize in {args.max_bullets} bullets with sentiment:\n\n{args.text}",
        schema=Summary,
        model="gpt-4o-mini"
    )
    return response

This demonstrates the platform's core pattern: - Input validation via Pydantic model - LLM integration with ctx.llm.complete() - Structured output enforced by schema parameter

2. Test Your Pack

huitzo pack dev
# Then in another terminal:
curl -X POST http://localhost:8080/api/v1/commands/math/add \
  -H "Content-Type: application/json" \
  -d '{"a": 2, "b": 3}'

Note: huitzo pack dev starts a local proxy on your machine. Your pack code executes in Huitzo's cloud sandbox—no local database or Redis required. See Developer Environment for details.

3. Deploy

huitzo pack build
huitzo pack publish

Core Concepts

Namespaces

Commands are organized in scoped namespaces using the format @{scope}/{pack-name}/{command}:

@acme/claims-processor/process-claim
@huitzo/core/hello

The WebCLI supports cd-style navigation between scopes and packs.

See: Namespaces Reference

Commands

A command is a single unit of work. It: - Has a unique name within a namespace - Takes validated input (Pydantic model) - Has access to platform services via Context - Returns a result or raises an error

See: Commands Reference

Context

The Context object is passed to every command and provides access to: - Storage (save/load data) - Integrations (LLM, email, files, etc.) - Metadata (user, tenant, correlation ID)

See: Context Reference

Storage

The storage interface provides tenant-isolated data persistence:

@command("save-note", namespace="notes")
async def save_note(args: NoteArgs, ctx: Context) -> dict:
    await ctx.storage.save(f"note:{args.id}", {"content": args.content})
    return {"saved": True}

See: Storage Reference

Integrations

Built-in integrations for common services:

  • ctx.llm – AI/LLM providers (OpenAI, Anthropic)
  • ctx.email – Send emails
  • ctx.files – Read/write files (Excel, CSV, etc.)
  • ctx.http – Make HTTP requests
  • ctx.telegram – Send Telegram messages

See: Integrations Reference

Custom Drivers (Enterprise)

For self-hosted Enterprise environments, custom tools may be available via ctx.drivers:

# Accessing a custom enterprise driver
result = await ctx.drivers.oracle_legacy.query("SELECT * FROM users")

See: Extensibility Architecture

Component Model (Design Complete)

The SDK's extensibility model is expanding beyond commands into a full methodology framework with four pillars: Components, Composition, Communication, and Configuration.

Component types: - HuitzoCommand — Unit of work (production) - HuitzoIntegration — Reusable service connector, accessed via ctx.integrations.<name> (design complete, implementation v2.5)

Future SDK extensions: - ctx.integrations — Pack-developer-created service connectors with lifecycle management - ctx.config (hierarchical) — Configuration cascade: platform -> pack -> command -> user - ctx.events — Async pack-to-pack communication via event bus - ctx.pipeline — Declarative command piping with streaming - ctx.files (VFS) — Mountable file system drivers

See: Component Model | Components Reference | Building Methodology

Command Patterns

Best for simple commands:

@command("greet", namespace="example")
async def greet(args: GreetArgs, ctx: Context) -> str:
    return f"Hello, {args.name}!"

Class Pattern

Best for complex commands with lifecycle hooks:

from huitzo_sdk import HuitzoCommand

class AnalyzeCommand(HuitzoCommand):
    name = "analyze"
    namespace = "data"

    def configure(self):
        self.timeout = 3600  # 1 hour
        self.retries = 5

    def on_progress(self, percent: int, message: str):
        self.update_state(progress=percent, status=message)

    async def execute(self, args: AnalyzeArgs, ctx: Context) -> dict:
        # Long-running analysis with progress updates
        for i, chunk in enumerate(data_chunks):
            await self.process(chunk)
            self.on_progress(i * 10, f"Processing chunk {i}")
        return {"status": "complete"}

Error Handling

The SDK provides a structured exception hierarchy:

from huitzo_sdk.errors import (
    HuitzoError,          # Base class for all Huitzo exceptions
    ValidationError,      # Input/output validation failed
    TimeoutError,         # Command timeout exceeded
    StorageError,         # Storage operation failed
    IntegrationError,     # External service failed (base)
    LLMError,             # LLM provider error
    EmailError,           # Email service error
    HTTPError,            # HTTP request failed
    PackExecutionError,   # Pack execution failed
    PermissionError,      # Insufficient permissions
    ConfigurationError,   # Invalid configuration
)

See Error Handling Reference for complete documentation including error codes, retry policies, and structured logging.

Example: Custom Error Handling

from huitzo_sdk.errors import CommandError

@command("risky", namespace="example")
async def risky(args: Args, ctx: Context) -> dict:
    try:
        result = await dangerous_operation()
        return result
    except SomeException as e:
        raise CommandError(
            message="Operation failed",
            command_name="risky",
            exit_code=1,
            details={"original_error": str(e)}
        )

Configuration

Per-Command Configuration

@command(
    "slow-task",
    namespace="tasks",
    timeout=3600,       # 1 hour
    retries=5,          # 5 retry attempts
    queue="long",       # Use long-running queue
)
async def slow_task(args: Args, ctx: Context) -> dict:
    ...

Environment Variables

The SDK automatically reads HUITZO_* environment variables:

Variable Purpose Default
HUITZO_LOG_LEVEL Logging level INFO
HUITZO_TIMEOUT Default timeout (seconds) 60
HUITZO_RETRIES Default retry count 3

Sync vs Async

The SDK supports both synchronous and asynchronous commands:

# Async (recommended for I/O operations)
@command("async-cmd", namespace="example")
async def async_cmd(args: Args, ctx: Context) -> dict:
    data = await ctx.http.get("https://api.example.com/data")
    return data

# Sync (for CPU-bound operations)
@command("sync-cmd", namespace="example")
def sync_cmd(args: Args, ctx: Context) -> dict:
    result = heavy_computation(args.data)
    return {"result": result}

Next Steps

  1. Create your first pack
  2. Understand the Context object
  3. Learn about storage
  4. Explore integrations