huitzo-sdk 1.0.0

Huitzo SDK for building Intelligence Packs

Huitzo SDK

The Huitzo SDK is a Python framework for building Intelligence Packs—reusable, scalable command modules that integrate with the Huitzo platform.

Features

• 🎯 Simple Command Decorator: Define commands with the @command decorator
• 🔌 Built-in Integrations: Pre-configured clients for LLM, Email, HTTP, Telegram, and File operations
• 💾 Flexible Storage: Pluggable storage backends with namespace isolation
• 🔄 Async First: Full async/await support with automatic sync function handling
• ✅ Type Safe: Built on Pydantic for runtime validation and type hints
• 🛡️ Error Handling: Comprehensive error hierarchy for robust error management

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()

Core Concepts

Commands

Commands are the building blocks of Intelligence Packs. Use the @command decorator to register a sync or async function as a command.

Decorator parameters:

• name: Command identifier
• namespace: Organizational grouping
• version: Semantic version
• timeout: Max execution time (default: 60s)
• retries: Number of retry attempts (default: 3)

Context

The Context object provides access to:

• Identity: user_id, tenant_id, session_id, correlation_id
• Metadata: command_name, namespace, command_version
• Integrations: llm, email, http, telegram, files

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

Integrations

Built-in clients for common operations:

• LLMClient: Language model completions and chat
• EmailClient: Send emails with templates
• HTTPClient: HTTP requests with security controls
• TelegramClient: Send messages and interact with Telegram
• FileClient: File operations and storage

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,
    TimeoutError,
    IntegrationError,
    LLMError,
    HTTPError,
    StorageError
)

try:
    result = await ctx.llm.complete(prompt="Hello")
except LLMError as e:
    # Handle LLM-specific errors
    print(f"LLM failed: {e}")
except TimeoutError as e:
    # Handle timeout errors
    print(f"Request timed out: {e}")
except HuitzoError as e:
    # Catch-all for other SDK errors
    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: [e-mail]
• 💬 Community: Discord
• 📚 Documentation: docs.huitzo.com
• 🐛 Issues: GitHub Issues

---

Built with ❤️ by the Huitzo team