Error Handling

sdk errors exceptions handling

Error Handling

This guide covers error handling patterns for Intelligence Pack development, including the SDK exception hierarchy, retry behavior, and best practices.

SDK Exception Hierarchy

The SDK provides a structured exception hierarchy for consistent error handling:

HuitzoError (base)
├── CommandError              # General command failure
├── ValidationError           # Input validation failures
├── CommandTimeoutError       # Command exceeded timeout
├── StorageError              # Storage operations failed
├── SecretsError              # User secret access failures
├── ExternalAPIError          # User-configured external API failures
├── IntegrationError          # Platform service failures
│   ├── IntegrationConnectionError  # (Proposed) Connection pool exhausted, DNS failure
│   ├── IntegrationAuthError        # (Proposed) Invalid credentials, expired token
│   ├── IntegrationTimeoutError     # (Proposed) External service did not respond
│   ├── IntegrationRateLimitError   # (Proposed) External service rate limit hit
│   ├── LLMError              # LLM provider errors
│   ├── CronError             # Cron scheduling errors
│   ├── EmailError            # Email sending errors
│   └── HTTPError             # HTTP request errors
├── SSHError                  # SSH execution or connection failures
├── MCPError                  # MCP server failures
│   ├── MCPConnectionError    # Server unreachable
│   ├── MCPToolError          # Tool execution failed
│   ├── MCPTimeoutError       # Timeout exceeded
│   └── MCPSchemaError        # Invalid arguments
├── PipelineError             # (Proposed) Pipeline execution failures
├── EventError                # (Proposed) Event bus failures
├── VFSError                  # (Proposed) Virtual file system failures
│   ├── MountError            # (Proposed) Mount point configuration failure
│   └── VFSFileNotFoundError  # (Proposed) File not found in mounted driver
├── PackExecutionError        # Cross-pack execution failed
├── PackPermissionError       # Insufficient permissions
├── ConfigurationError        # Invalid configuration
└── RateLimitError            # Rate limit exceeded

(Proposed) entries are architecturally specified but not yet implemented. See Component Model, Piping Protocol, Event Bus, and Virtual File System.

Importing Exceptions

from huitzo_sdk.errors import (
    HuitzoError,
    CommandError,
    ValidationError,
    CommandTimeoutError,
    StorageError,
    SecretsError,
    ExternalAPIError,
    IntegrationError,
    LLMError,
    CronError,
    EmailError,
    HTTPError,
    SSHError,
    MCPError,
    MCPConnectionError,
    MCPToolError,
    MCPTimeoutError,
    MCPSchemaError,
    PackExecutionError,
    PackPermissionError,
    ConfigurationError,
    RateLimitError,
)

Exception Reference

ValidationError

Raised when command input validation fails.

from huitzo_sdk.errors import ValidationError

@command("process", namespace="data")
async def process(args: ProcessArgs, ctx: Context) -> dict:
    if args.count < 0:
        raise ValidationError(
            field="count",
            value=args.count,
            message="Count must be non-negative"
        )

    if args.start_date > args.end_date:
        raise ValidationError(
            field="date_range",
            value=f"{args.start_date} - {args.end_date}",
            message="Start date must be before end date"
        )

Attributes: - field - The field that failed validation - value - The invalid value - message - Human-readable error message

CommandError

Raised when a command fails due to a general execution error. This is the catch-all error for command failures that don't fit into more specific categories.

from huitzo_sdk.errors import CommandError

@command("execute", namespace="tasks")
async def execute(args: Args, ctx: Context) -> dict:
    try:
        result = await perform_action()
    except SomeLibraryError as e:
        raise CommandError(
            message=f"Failed to execute action: {e}",
            exit_code=1,
            details={"original_error": str(e)}
        )
    return result

Attributes: - message - Human-readable error message - exit_code - Optional exit code for CLI contexts (default: 1) - details - Optional dictionary of additional context

CommandTimeoutError

Raised when a command exceeds its configured timeout.

from huitzo_sdk.errors import CommandTimeoutError

# SDK raises this automatically, but you can catch it:
@command("long-task", namespace="tasks", timeout=60)
async def long_task(args: Args, ctx: Context) -> dict:
    try:
        result = await slow_operation()
    except CommandTimeoutError:
        # Save partial progress
        await ctx.storage.save("partial-result", partial_data)
        raise  # Re-raise for platform handling

Attributes: - timeout_seconds - The configured timeout - elapsed_seconds - Actual elapsed time

StorageError

Raised when storage operations fail.

from huitzo_sdk.errors import StorageError

@command("save-data", namespace="data")
async def save_data(args: Args, ctx: Context) -> dict:
    try:
        await ctx.storage.save("key", large_data)
    except StorageError as e:
        ctx.log.error(f"Storage failed: {e.message}")
        raise

Attributes: - operation - The failed operation (save, get, delete, list) - key - The storage key involved - message - Error details

SecretsError

Raised when a required user secret is missing or inaccessible. User secrets are per-user API keys declared in the pack manifest and provided by end users.

from huitzo_sdk.errors import SecretsError

@command("sync-data", namespace="finance")
async def sync_data(args: Args, ctx: Context) -> dict:
    try:
        api_key = ctx.secrets.require("FINANCIAL_API_KEY")
    except SecretsError as e:
        # Return helpful guidance to the user
        return {
            "error": "Missing API key",
            "help": "Configure your API key in Settings → Pack Secrets",
            "secret_name": e.secret_name
        }

    # Use the secret...
    return await fetch_financial_data(api_key)

Attributes: - secret_name - The name of the missing secret - message - Human-readable error message

Related: See Secrets Management for the three-tier secrets model.

ExternalAPIError

Raised when an external API call fails due to invalid credentials or API errors. Use this for user-configured external services (accessed via user secrets) to provide clear feedback.

from huitzo_sdk.errors import ExternalAPIError

@command("fetch-accounts", namespace="finance")
async def fetch_accounts(args: Args, ctx: Context) -> dict:
    api_key = ctx.secrets.require("PLAID_API_KEY")

    try:
        response = await ctx.http.get(
            "https://api.plaid.com/accounts",
            headers={"Authorization": f"Bearer {api_key}"}
        )
    except HTTPError as e:
        if e.status_code == 401:
            raise ExternalAPIError(
                service="plaid",
                message="Invalid API key. Verify your key in Settings → Pack Secrets."
            )
        if e.status_code == 403:
            raise ExternalAPIError(
                service="plaid",
                message="API key lacks required permissions. Check your Plaid dashboard."
            )
        raise

    return response

Attributes: - service - The external service name (e.g., "plaid", "stripe") - message - User-friendly error message with guidance

When to Use: - User's API key is invalid or expired - User's API key lacks required permissions - External service rate limits user's account - External service returns authentication errors

Related: See Secrets Management for user secrets patterns.

IntegrationError

Base class for external service failures. Use specific subclasses when possible.

from huitzo_sdk.errors import IntegrationError, LLMError, HTTPError

@command("analyze", namespace="ai")
async def analyze(args: Args, ctx: Context) -> dict:
    try:
        response = await ctx.llm.complete(prompt)
    except LLMError as e:
        ctx.log.error(f"LLM failed: {e.provider} - {e.message}")
        raise IntegrationError(
            service="analysis",
            message="AI analysis unavailable"
        )

LLMError Attributes: - provider - LLM provider (openai, anthropic) - model - Model that failed - status_code - HTTP status code if applicable

HTTPError Attributes: - url - Request URL - method - HTTP method - status_code - Response status code - response_body - Response body (truncated)

SSHError

Raised when an SSH connection or command execution fails. This covers connection failures, host key mismatches, target not found, command validation errors, and timeouts.

from huitzo_sdk.errors import SSHError

@command("run-task", namespace="compute")
async def run_task(args: Args, ctx: Context) -> dict:
    try:
        result = await ctx.ssh.run("gpu-cluster", args.command, timeout=60)
        return {"output": result.stdout, "exit_code": result.exit_code}
    except SSHError as e:
        ctx.log.error(f"SSH failed on {e.target}: {e.message}")
        return {"error": e.message, "target": e.target}

Attributes: - host - Hostname or IP of the target (may be empty if target not resolved) - target - Name of the SSH target that was requested - message - Human-readable error description

Common Causes: - Target not in pack's allowed list (ssh_targets.allowed in manifest) - Target not registered by the user - No host key fingerprint stored (needs /verify first) - Connection refused, timeout, or authentication failure - Command exceeds 4,096 byte limit or contains null bytes

Error Properties: - code = "SSH_ERROR" - http_status = 502 - retryable = True

Related: See SSH Integration for ctx.ssh API documentation.

MCPError

Base class for MCP (Model Context Protocol) server failures. Use specific subclasses for more precise error handling.

from huitzo_sdk.errors import (
    MCPError,
    MCPConnectionError,
    MCPToolError,
    MCPTimeoutError,
    MCPSchemaError
)

@command("github-issue", namespace="devtools")
async def github_issue(args: Args, ctx: Context) -> dict:
    try:
        result = await ctx.mcp.call("github", "create_issue", {
            "owner": args.owner,
            "repo": args.repo,
            "title": args.title
        })
        return {"issue": result}

    except MCPConnectionError as e:
        ctx.log.error(f"GitHub MCP server unavailable: {e.message}")
        return {"error": "GitHub integration temporarily unavailable", "retry": True}

    except MCPSchemaError as e:
        ctx.log.error(f"Invalid arguments: {e.message}")
        return {"error": f"Invalid input: {e.field}", "validation_errors": e.errors}

    except MCPToolError as e:
        ctx.log.error(f"GitHub tool failed: {e.message}")
        return {"error": f"GitHub error: {e.message}", "code": e.code}

    except MCPTimeoutError as e:
        ctx.log.error(f"GitHub call timed out after {e.timeout}s")
        return {"error": "Request timed out", "retry": True}

MCPConnectionError

Raised when the MCP server cannot be reached or crashes.

Attributes: - server - Name of the MCP server that failed to connect - message - Human-readable error description

Common Causes: - Server process crashed or failed to start - Network issues (for HTTP transport) - Invalid command configuration (for STDIO transport) - Missing or invalid environment variables

MCPToolError

Raised when an MCP tool execution fails.

Attributes: - server - MCP server name - tool - Tool name that failed - code - Error code from the MCP server - message - Error message from the tool

Common Causes: - Tool returned an error (e.g., GitHub API returned 404) - Tool doesn't exist on the server - Permission denied by external service

MCPTimeoutError

Raised when an MCP tool execution exceeds the configured timeout.

Attributes: - server - MCP server name - tool - Tool name that timed out - timeout - Configured timeout value in seconds

Solutions: - Increase timeout in the ctx.mcp.call() call - Check if the operation is genuinely slow - Consider breaking into smaller operations

MCPSchemaError

Raised when arguments don't match the tool's JSON Schema.

Attributes: - server - MCP server name - tool - Tool name - field - Field that failed validation - errors - List of validation error details - message - Human-readable validation message

Common Causes: - Missing required field - Wrong field type (e.g., string instead of number) - Value doesn't match pattern or constraints

Related: See MCP Reference for complete MCP API documentation.

PackExecutionError

Raised when cross-pack command execution fails.

from huitzo_sdk.errors import PackExecutionError

@command("workflow", namespace="automation")
async def workflow(args: Args, ctx: Context) -> dict:
    try:
        result = await ctx.execute("other-pack", "command", args={})
    except PackExecutionError as e:
        ctx.log.error(f"Cross-pack call failed: {e.pack}.{e.command}")
        ctx.log.error(f"Original error: {e.original_error}")
        raise

Attributes: - pack - Target pack namespace - command - Target command name - original_error - The underlying exception

RateLimitError

Raised when a rate limit is exceeded. The platform or external services may impose rate limits on command execution.

from huitzo_sdk.errors import RateLimitError

@command("bulk-process", namespace="data")
async def bulk_process(args: Args, ctx: Context) -> dict:
    try:
        result = await process_batch(args.items)
    except RateLimitError as e:
        ctx.log.warning(
            f"Rate limit hit. Retry after {e.retry_after}s",
            extra={"limit": e.limit, "current": e.current}
        )
        # Return partial results or schedule for later
        return {
            "status": "rate_limited",
            "retry_after": e.retry_after,
            "processed": partial_results
        }

Attributes: - retry_after - Seconds to wait before retrying (from rate limit header) - limit - The rate limit that was exceeded (e.g., "100/hour") - current - Current usage count when limit was hit

Note: The platform automatically handles RateLimitError with exponential backoff for retryable commands. See Automatic Retry Behavior below.

Automatic Retry Behavior

The SDK automatically retries commands on certain transient errors:

Retryable Errors

Error Type Retried Max Attempts Backoff
Network timeout ✅ 3 Exponential
Connection reset ✅ 3 Exponential
Rate limit (429) ✅ 5 Rate limit header
Server error (5xx) ✅ 3 Exponential
RateLimitError ✅ 5 Rate limit header
ValidationError ❌ - -
CommandError ❌ - -
PackPermissionError ❌ - -
ConfigurationError ❌ - -

Configuring Retries

@command(
    "api-call",
    namespace="external",
    retries=5,           # Max retry attempts
    retry_backoff=2.0,   # Backoff multiplier
    retry_max_wait=60,   # Max wait between retries (seconds)
)
async def api_call(args: Args, ctx: Context) -> dict:
    ...

Disabling Retries

@command("no-retry", namespace="critical", retries=0)
async def no_retry(args: Args, ctx: Context) -> dict:
    # This command will not be retried on failure
    ...

Error Response Format

When a command fails, both the production backend and the sandbox dev server return a structured error response. The sandbox proxy mirrors the same format so developers see consistent error handling across environments:

{
  "success": false,
  "error": {
    "type": "ValidationError",
    "message": "Count must be non-negative",
    "code": "VALIDATION_FAILED",
    "details": {
      "field": "count",
      "value": -5
    },
    "correlation_id": "abc-123-def",
    "timestamp": "2026-01-22T10:30:00Z"
  }
}

Error Codes

Code HTTP Status Description
COMMAND_FAILED 500 General command execution failure
VALIDATION_FAILED 400 Input validation error
SECRET_MISSING 400 Required user secret not configured
PERMISSION_DENIED 403 Insufficient permissions
NOT_FOUND 404 Resource not found
TIMEOUT 408 Command timeout exceeded
RATE_LIMITED 429 Rate limit exceeded
INTERNAL_ERROR 500 Unexpected server error
EXTERNAL_API_ERROR 502 User-configured external API failure
SSH_ERROR 502 SSH connection or execution failure
INTEGRATION_ERROR 502 Platform service failure
SERVICE_UNAVAILABLE 503 Service temporarily unavailable

Logging Errors with Correlation IDs

Always use the context logger to ensure correlation IDs are included:

@command("example", namespace="demo")
async def example(args: Args, ctx: Context) -> dict:
    try:
        result = await risky_operation()
    except Exception as e:
        # ✅ Good: Uses context logger (includes correlation_id)
        ctx.log.error(
            "Operation failed",
            extra={
                "error_type": type(e).__name__,
                "error_message": str(e),
                "args": args.model_dump(),
            }
        )
        raise

    return result

Log output includes correlation ID automatically:

{
  "level": "ERROR",
  "message": "Operation failed",
  "correlation_id": "abc-123-def",
  "tenant_id": "org_456",
  "user_id": "user_789",
  "command": "demo.example",
  "error_type": "ValueError",
  "error_message": "Invalid input",
  "timestamp": "2026-01-22T10:30:00Z"
}

Best Practices

1. Be Specific with Exceptions

# ✅ Good: Specific exception with context
raise ValidationError(
    field="email",
    value=args.email,
    message="Invalid email format"
)

# ❌ Bad: Generic exception
raise Exception("Bad email")

2. Include Actionable Information

# ✅ Good: Tells user what to do
raise ValidationError(
    field="file_size",
    value=f"{size_mb}MB",
    message=f"File size ({size_mb}MB) exceeds maximum of 10MB. Please compress or split the file."
)

# ❌ Bad: No guidance
raise ValidationError(field="file_size", value=size_mb, message="Too big")

3. Clean Up on Errors

@command("process-file", namespace="files")
async def process_file(args: Args, ctx: Context) -> dict:
    temp_file = None
    try:
        temp_file = await download_file(args.url)
        result = await process(temp_file)
        return result
    except Exception:
        ctx.log.error("Processing failed, cleaning up")
        raise
    finally:
        if temp_file:
            await cleanup(temp_file)

4. Don't Swallow Errors Silently

# ✅ Good: Log and handle appropriately
try:
    result = await external_api.call()
except HTTPError as e:
    ctx.log.warning(f"API call failed: {e.status_code}")
    return {"status": "degraded", "cached": True, "data": cached_data}

# ❌ Bad: Silent failure
try:
    result = await external_api.call()
except:
    pass  # Never do this!

5. Use Structured Error Data

# ✅ Good: Structured error with details
raise IntegrationError(
    service="payment-provider",
    message="Payment declined",
    details={
        "provider_code": "CARD_DECLINED",
        "last_four": "4242",
        "retry_allowed": True,
    }
)

Custom Error Types

Pack developers can extend the SDK error hierarchy to create domain-specific errors. Custom errors integrate with the platform's error handling, logging, and response formatting.

Extending HuitzoError

Subclass HuitzoError (or any SDK error) to create domain-specific errors. The pattern:

  1. Set class attributes — code, http_status, retryable
  2. Add domain context — extra fields relevant to your pack's domain
  3. Chain exceptions — use raise ... from e to preserve the original cause
# pseudocode — custom error pattern

class YourPackError(HuitzoError):
    # Set a machine-readable error code
    code = "YOUR_PACK_ERROR"
    # Set the HTTP status code for API responses
    http_status = 400

    # Add domain-specific context fields
    # e.g., record_id, field_name, reason, etc.
    ...

class YourPackRetryableError(YourPackError):
    code = "YOUR_PACK_RETRYABLE"
    http_status = 500
    retryable = True   # Platform will automatically retry
    ...

Using Custom Errors

# pseudocode — raising custom errors in a command

@command("your-action", namespace="your-pack")
async def your_action(args, ctx):
    # Validate input → raise domain-specific validation error
    if invalid(args.some_field):
        raise YourPackError(message="...", field="some_field", reason="...")

    # Wrap integration failures → chain the original exception
    try:
        result = await ctx.llm.complete(...)
    except LLMError as e:
        raise YourPackRetryableError(message="...", details={...}) from e

    return result

Error Response Format

Custom errors are serialized with all attributes:

{
  "success": false,
  "error": {
    "type": "ClaimValidationError",
    "message": "Invalid policy number format",
    "code": "CLAIM_VALIDATION_FAILED",
    "details": {
      "claim_id": "CLM-12345",
      "field": "policy_number",
      "reason": "Must start with 'POL-'"
    },
    "correlation_id": "abc-123-def",
    "timestamp": "2026-01-22T10:30:00Z"
  }
}

Custom Error Attributes

Attribute Type Purpose
code str Machine-readable error code (e.g., CLAIM_VALIDATION_FAILED)
http_status int HTTP status code for API responses (default: 500)
retryable bool Whether the platform should retry on this error (default: False)
details dict Additional context included in error response

Extending Integration Errors

For external service integrations, extend IntegrationError:

from huitzo_sdk.errors import IntegrationError

class InsuranceAPIError(IntegrationError):
    """Insurance provider API failed."""

    code = "INSURANCE_API_ERROR"
    http_status = 502
    retryable = True

    def __init__(
        self,
        message: str,
        provider: str,
        endpoint: str,
        status_code: int | None = None,
        **kwargs
    ):
        super().__init__(
            service=f"insurance:{provider}",
            message=message,
            **kwargs
        )
        self.provider = provider
        self.endpoint = endpoint
        self.status_code = status_code

Error Hierarchy Best Practices

  1. Create a base error for your pack (e.g., ClaimsError)
  2. Extend for specifics – validation, processing, external services
  3. Include context – IDs, field names, actionable information
  4. Set appropriate codes – consistent, machine-readable
  5. Mark retryable – only for transient failures
ClaimsError (pack base)
├── ClaimValidationError      # Input validation
├── ClaimProcessingError      # Processing failures
├── PolicyLookupError         # Policy not found
└── InsuranceAPIError         # External API failures

Testing Error Handling

import pytest
from huitzo_sdk.errors import ValidationError
from huitzo_sdk.testing import MockContext

async def test_validation_error():
    ctx = MockContext()

    with pytest.raises(ValidationError) as exc_info:
        await my_command(InvalidArgs(count=-1), ctx)

    assert exc_info.value.field == "count"
    assert "non-negative" in exc_info.value.message

async def test_error_logging():
    ctx = MockContext()

    with pytest.raises(SomeError):
        await my_command(args, ctx)

    # Verify error was logged
    assert ctx.log.error.called
    assert "correlation_id" in ctx.log.error.call_args


async def test_custom_error():
    """Test custom pack errors."""
    from my_pack.errors import ClaimValidationError

    ctx = MockContext()

    with pytest.raises(ClaimValidationError) as exc_info:
        await process_claim(InvalidClaimArgs(policy_number="INVALID"), ctx)

    error = exc_info.value
    assert error.claim_id == "CLM-12345"
    assert error.field == "policy_number"
    assert error.code == "CLAIM_VALIDATION_FAILED"