Huitzo SDK

A Python framework for building modular command functions that integrate with the Huitzo platform.

Features

• Simple command decorator for defining functions
• Built-in clients for LLM, Email, HTTP, Telegram, SSH, Database, and File operations
• Pluggable storage backends with namespace isolation
• Full async/await support with automatic sync function handling
• Type-safe with Pydantic validation
• Comprehensive error handling

Installation

pip install huitzo-sdk

Quick Start

Defining a Command

from pydantic import BaseModel
from huitzo_sdk import command, Context, Result

class GreetingArgs(BaseModel):
    name: str
    language: str = "en"

@command(
    name="greet",
    namespace="examples",
    version="1.0.0",
    description="Greet a user in their language"
)
async def greet_user(args: GreetingArgs, ctx: Context) -> Result:
    greetings = {"en": "Hello", "es": "Hola", "fr": "Bonjour"}
    greeting = greetings.get(args.language, "Hello")
    
    return {
        "message": f"{greeting}, {args.name}!",
        "user_id": str(ctx.user_id)
    }

Using Built-in Integrations

from pydantic import BaseModel
from huitzo_sdk import command, Context, Result

class WeatherArgs(BaseModel):
    city: str

@command(
    name="get_weather",
    namespace="weather",
    version="1.0.0",
)
async def get_weather(args: WeatherArgs, ctx: Context) -> Result:
    # Use built-in HTTP client
    response = await ctx.http.get(
        f"https://api.weather.com/v1/forecast?city={args.city}"
    )
    return response.json()

Complete Example: Data Processing Pipeline

from pydantic import BaseModel
from huitzo_sdk import command, Context, Result

class ProcessDataArgs(BaseModel):
    data_url: str
    recipient_email: str

@command(
    name="process_data",
    namespace="analytics",
    version="1.0.0",
    description="Fetch, analyze, and report on data"
)
async def process_data(args: ProcessDataArgs, ctx: Context) -> Result:
    # Fetch data from external API
    response = await ctx.http.get(args.data_url)
    data = response.json()
    
    # Store raw data
    await ctx.storage.set(f"raw_data:{str(ctx.session_id)}", data)
    
    # Analyze with LLM (limit data size to avoid token limits)
    data_summary = str(data)[:500]
    analysis = await ctx.llm.complete(
        prompt=f"Analyze this data and provide insights: {data_summary}",
        model="gpt-4"
    )
    
    # Send results via email
    await ctx.email.send(
        to=args.recipient_email,
        subject="Data Analysis Complete",
        body=f"Analysis Results:\n\n{analysis}"
    )
    
    return {
        "status": "complete",
        "session_id": str(ctx.session_id),
        "data_points": len(data)
    }

Core Concepts

Commands

Commands are functions decorated with @command. The decorator supports both sync and async functions.

Decorator parameters:

• name: Command identifier
• namespace: Organizational grouping
• version: Semantic version
• timeout: Maximum execution time in seconds (default: 60)
• retries: Number of retry attempts (default: 3)
• description: Optional command description

Context

The Context object provides access to:

• Identity: user_id, tenant_id, session_id, correlation_id
• Metadata: command_name, namespace, command_version
• Integration Clients: llm, email, http, telegram, ssh, db, files
• Platform Services: storage, secrets, log, cron

async def my_command(args: dict, ctx: Context) -> Result:
    # Access user identity
    user_id = ctx.user_id
    
    # Use integration clients
    completion = await ctx.llm.complete(
        prompt="Explain quantum computing",
        model="gpt-4"
    )
    
    # Send results via email
    await ctx.email.send(
        to="user@example.com",
        subject="Your Results",
        body=completion
    )
    
    return {"status": "sent"}

Integration Clients

Built-in clients for external services:

• LLMClient: Language model completions and chat
• EmailClient: Send emails with templates
• HTTPClient: HTTP requests with security controls
• TelegramClient: Send messages and interact with Telegram
• SSHClient: Execute remote commands via SSH
• DBClient: Database operations with transaction support
• FileClient: File operations and storage

Platform Services

Built-in platform utilities:

• StorageClient: Persistent key-value storage with namespace isolation
• SecretsClient: Secure credential management
• LogClient: Structured logging
• CronClient: Schedule recurring tasks

Storage

Persistent storage with namespace isolation:

from huitzo_sdk.storage import InMemoryBackend, StorageNamespace

# Create storage backend
backend = InMemoryBackend()
storage = StorageNamespace(backend=backend, namespace="my-app")

# Store and retrieve data
await storage.set("user:123", {"name": "Alice", "score": 100})
user_data = await storage.get("user:123")

Error Handling

The SDK provides a comprehensive error hierarchy:

from huitzo_sdk import (
    HuitzoError,
    ValidationError,
    CommandTimeoutError,
    IntegrationError,
    LLMError,
    HTTPError,
    StorageError
)

try:
    result = await ctx.llm.complete(prompt="Hello")
except LLMError as e:
    print(f"LLM failed: {e}")
except CommandTimeoutError as e:
    print(f"Request timed out: {e}")
except HuitzoError as e:
    print(f"Huitzo error: {e}")

Development

Prerequisites

• Python 3.11 or higher
• uv (recommended) or pip

Setup

# Clone the repository
git clone https://github.com/Huitzo-Inc/sdk.git
cd sdk

# Install dependencies
uv sync --dev

# Or with pip
pip install -e ".[dev]"

Running Tests

# Run all tests
pytest

# Run with coverage
pytest --cov=huitzo_sdk --cov-report=term-missing

# Run specific test file
pytest tests/test_command.py

Code Quality

# Type checking
mypy src/huitzo_sdk

# Linting and formatting
ruff check src tests
ruff format src tests

Project Structure

huitzo-sdk/
├── src/
│   └── huitzo_sdk/
│       ├── __init__.py       # Public API exports
│       ├── command.py        # @command decorator
│       ├── context.py        # Context object
│       ├── types.py          # Core type definitions
│       ├── errors.py         # Error hierarchy
│       ├── integrations/     # Integration clients
│       │   ├── llm.py
│       │   ├── email.py
│       │   ├── http.py
│       │   ├── telegram.py
│       │   └── files.py
│       └── storage/          # Storage backends
│           ├── protocol.py
│           ├── memory.py
│           └── namespace.py
├── tests/                    # Test suite
├── pyproject.toml           # Project configuration
└── README.md                # This file

Contributing

We welcome contributions! Please:

1. Fork the repository
2. Create a feature branch (git checkout -b feature/amazing-feature)
3. Make your changes with tests
4. Ensure all tests pass (pytest)
5. Run linters (ruff check src tests)
6. Commit your changes (git commit -m 'Add amazing feature')
7. Push to the branch (git push origin feature/amazing-feature)
8. Open a Pull Request

License

See the LICENSE file for details.

Support

• Email: support@huitzo.com
• Community: Discord
• Documentation: docs.huitzo.com
• Issues: GitHub Issues