External Integrations

integration api external sdk

External Integrations

This guide covers how to integrate external applications with Huitzo without using the Dashboard SDK. Use this approach when building custom applications in any language or framework.

Overview

External applications interact with Huitzo through the REST API:

┌─────────────────────────────────────────────────────────────────┐
│                    External Application                          │
│                                                                  │
│  ┌──────────────────────────────────────────────────────────┐   │
│  │   Your App (Any Language/Framework)                       │   │
│  │   • Mobile App (iOS, Android)                            │   │
│  │   • Desktop App (Electron, native)                       │   │
│  │   • Backend Service (Python, Go, Node.js)                │   │
│  │   • Legacy System Integration                            │   │
│  └────────────────────────┬─────────────────────────────────┘   │
│                           │                                      │
│                    REST API Calls                                │
│                           │                                      │
└───────────────────────────┼──────────────────────────────────────┘
                            │
                            ▼
┌───────────────────────────────────────────────────────────────────┐
│                        Huitzo API                                  │
│                  https://huitzo.ai                            │
│                                                                    │
│  • Authentication (OAuth 2.0, API Keys)                           │
│  • Command Execution                                              │
│  • WebSocket for Real-Time Updates                                │
└───────────────────────────────────────────────────────────────────┘

Authentication

Use OAuth when your app needs to act on behalf of a user:

1. Redirect user to: https://auth.huitzo.com/authorize
   ?client_id=YOUR_CLIENT_ID
   &redirect_uri=YOUR_REDIRECT_URI
   &response_type=code
   &scope=commands:execute

2. User authenticates with Huitzo

3. Receive authorization code at redirect_uri

4. Exchange code for tokens:
   POST https://auth.huitzo.com/token
   {
     "grant_type": "authorization_code",
     "code": "AUTH_CODE",
     "client_id": "YOUR_CLIENT_ID",
     "client_secret": "YOUR_CLIENT_SECRET",
     "redirect_uri": "YOUR_REDIRECT_URI"
   }

5. Use access_token for API calls

Option 2: API Keys (For Server-to-Server)

Use API keys for backend integrations where no user interaction is needed:

# Get an API key from Dashboard → Settings → API Keys

# Use in requests
curl -X POST https://huitzo.ai/api/v1/commands/@acme/claims/process \
  -H "Authorization: Bearer sk_live_xxxxx" \
  -H "Content-Type: application/json" \
  -d '{"claimId": "CLM-12345"}'

Token Refresh

Access tokens expire after 30 minutes. Use refresh tokens to get new access tokens:

POST https://auth.huitzo.com/token
Content-Type: application/json

{
  "grant_type": "refresh_token",
  "refresh_token": "YOUR_REFRESH_TOKEN",
  "client_id": "YOUR_CLIENT_ID"
}

Executing Commands

Basic Command Execution

POST /api/v1/commands/{scope}/{pack}/{command}

Example:

curl -X POST https://huitzo.ai/api/v1/commands/@acme/claims/process-claim \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "claimId": "CLM-12345",
    "priority": "high"
  }'

Response:

{
  "success": true,
  "data": {
    "id": "CLM-12345",
    "status": "approved",
    "processedAt": "2026-01-24T10:30:00Z"
  },
  "executionTime": 1234,
  "correlationId": "abc-123-def"
}

Async Command Execution

For long-running commands, use async mode:

POST /api/v1/commands/@acme/reports/generate?async=true

Response:

{
  "taskId": "task_abc123",
  "status": "pending",
  "statusUrl": "/api/v1/tasks/task_abc123"
}

Poll for completion:

GET /api/v1/tasks/task_abc123

Real-Time Updates

WebSocket Connection

Connect to WebSocket for real-time events:

const ws = new WebSocket('wss://huitzo.ai/ws');

// Authenticate
ws.send(JSON.stringify({
  type: 'authenticate',
  token: 'YOUR_ACCESS_TOKEN'
}));

// Subscribe to events
ws.send(JSON.stringify({
  type: 'subscribe',
  channels: ['claims:updated', 'claims:created']
}));

// Handle messages
ws.onmessage = (event) => {
  const data = JSON.parse(event.data);
  console.log('Event:', data);
};

Event Types

Event Description Payload
{entity}:created New item created { id, type, data }
{entity}:updated Item updated { id, type, changes }
{entity}:deleted Item deleted { id, type }
command:started Command execution started { correlationId, command }
command:completed Command completed { correlationId, result }
command:failed Command failed { correlationId, error }

Language Examples

Python

import httpx

class HuitzoClient:
    def __init__(self, api_key: str, base_url: str = "https://huitzo.ai"):
        self.api_key = api_key
        self.base_url = base_url
        self.client = httpx.AsyncClient(
            base_url=base_url,
            headers={"Authorization": f"Bearer {api_key}"}
        )

    async def execute(self, command: str, args: dict) -> dict:
        """Execute a command.

        Args:
            command: Full command path (e.g., "@acme/claims/process")
            args: Command arguments

        Returns:
            Command result
        """
        response = await self.client.post(
            f"/api/v1/commands/{command}",
            json=args
        )
        response.raise_for_status()
        return response.json()

# Usage
async def main():
    huitzo = HuitzoClient(api_key="sk_live_xxxxx")

    result = await huitzo.execute(
        "@acme/claims/process-claim",
        {"claimId": "CLM-12345"}
    )
    print(f"Claim status: {result['data']['status']}")

Node.js / TypeScript

import axios, { AxiosInstance } from 'axios';

class HuitzoClient {
  private client: AxiosInstance;

  constructor(apiKey: string, baseUrl = 'https://huitzo.ai') {
    this.client = axios.create({
      baseURL: baseUrl,
      headers: {
        'Authorization': `Bearer ${apiKey}`,
        'Content-Type': 'application/json',
      },
    });
  }

  async execute<T>(command: string, args: object): Promise<T> {
    const response = await this.client.post(`/api/v1/commands/${command}`, args);
    return response.data;
  }
}

// Usage
const huitzo = new HuitzoClient('sk_live_xxxxx');

interface ClaimResult {
  id: string;
  status: 'approved' | 'denied' | 'pending';
}

const result = await huitzo.execute<ClaimResult>(
  '@acme/claims/process-claim',
  { claimId: 'CLM-12345' }
);
console.log(`Claim status: ${result.status}`);

Go

package huitzo

import (
    "bytes"
    "encoding/json"
    "fmt"
    "net/http"
)

type Client struct {
    APIKey  string
    BaseURL string
    HTTP    *http.Client
}

func NewClient(apiKey string) *Client {
    return &Client{
        APIKey:  apiKey,
        BaseURL: "https://huitzo.ai",
        HTTP:    &http.Client{},
    }
}

type CommandResult struct {
    Success       bool            `json:"success"`
    Data          json.RawMessage `json:"data"`
    ExecutionTime int             `json:"executionTime"`
    CorrelationID string          `json:"correlationId"`
}

func (c *Client) Execute(command string, args interface{}) (*CommandResult, error) {
    body, err := json.Marshal(args)
    if err != nil {
        return nil, err
    }

    req, err := http.NewRequest(
        "POST",
        fmt.Sprintf("%s/api/v1/commands/%s", c.BaseURL, command),
        bytes.NewBuffer(body),
    )
    if err != nil {
        return nil, err
    }

    req.Header.Set("Authorization", "Bearer "+c.APIKey)
    req.Header.Set("Content-Type", "application/json")

    resp, err := c.HTTP.Do(req)
    if err != nil {
        return nil, err
    }
    defer resp.Body.Close()

    var result CommandResult
    if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
        return nil, err
    }

    return &result, nil
}

// Usage
func main() {
    client := NewClient("sk_live_xxxxx")

    result, err := client.Execute("@acme/claims/process-claim", map[string]string{
        "claimId": "CLM-12345",
    })
    if err != nil {
        panic(err)
    }

    fmt.Printf("Success: %v\n", result.Success)
}

cURL

# Execute a command
curl -X POST https://huitzo.ai/api/v1/commands/@acme/claims/process-claim \
  -H "Authorization: Bearer sk_live_xxxxx" \
  -H "Content-Type: application/json" \
  -d '{"claimId": "CLM-12345"}'

# List available commands
curl https://huitzo.ai/api/v1/commands \
  -H "Authorization: Bearer sk_live_xxxxx"

# Get command details
curl https://huitzo.ai/api/v1/commands/@acme/claims/process-claim \
  -H "Authorization: Bearer sk_live_xxxxx"

Error Handling

Error Response Format

{
  "success": false,
  "error": {
    "type": "ValidationError",
    "message": "Invalid claim ID format",
    "code": "VALIDATION_FAILED",
    "details": {
      "field": "claimId",
      "value": "invalid"
    },
    "correlationId": "abc-123-def"
  }
}

Error Codes

Code HTTP Status Description
VALIDATION_FAILED 400 Input validation error
UNAUTHORIZED 401 Invalid or missing token
FORBIDDEN 403 Insufficient permissions
NOT_FOUND 404 Command or resource not found
RATE_LIMITED 429 Too many requests
INTERNAL_ERROR 500 Server error

Retry Strategy

Implement exponential backoff for transient errors:

import time
import random

def execute_with_retry(client, command, args, max_retries=3):
    for attempt in range(max_retries):
        try:
            return client.execute(command, args)
        except RateLimitError:
            if attempt == max_retries - 1:
                raise
            wait = (2 ** attempt) + random.uniform(0, 1)
            time.sleep(wait)
        except ServerError:
            if attempt == max_retries - 1:
                raise
            wait = (2 ** attempt) + random.uniform(0, 1)
            time.sleep(wait)

Rate Limits

Limit Value Scope
Requests per minute 100 Per API key
Requests per hour 1000 Per API key
Concurrent connections 10 Per API key
WebSocket messages 100/min Per connection

Rate limit headers are included in responses:

X-RateLimit-Limit: 100
X-RateLimit-Remaining: 95
X-RateLimit-Reset: 1706088000

Webhooks

Register webhooks to receive events at your endpoint:

Create Webhook

POST /api/v1/webhooks
{
  "url": "https://your-app.com/webhooks/huitzo",
  "events": ["claims:created", "claims:updated"],
  "secret": "whsec_your_secret"
}

Webhook Payload

{
  "id": "evt_abc123",
  "type": "claims:created",
  "timestamp": "2026-01-24T10:30:00Z",
  "data": {
    "id": "CLM-12345",
    "status": "pending"
  }
}

Verify Webhook Signature

import hmac
import hashlib

def verify_webhook(payload: bytes, signature: str, secret: str) -> bool:
    expected = hmac.new(
        secret.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(f"sha256={expected}", signature)

# In your webhook handler
@app.post("/webhooks/huitzo")
async def handle_webhook(request: Request):
    payload = await request.body()
    signature = request.headers.get("X-Huitzo-Signature")

    if not verify_webhook(payload, signature, WEBHOOK_SECRET):
        raise HTTPException(401, "Invalid signature")

    event = json.loads(payload)
    # Process event...

Best Practices

1. Use API Keys Securely

# ✅ Good: Load from environment
import os
api_key = os.environ["HUITZO_API_KEY"]

# ❌ Bad: Hardcoded
api_key = "sk_live_xxxxx"

2. Handle Token Expiration

class HuitzoClient:
    def __init__(self, client_id, client_secret):
        self.client_id = client_id
        self.client_secret = client_secret
        self.access_token = None
        self.token_expires_at = None

    async def ensure_token(self):
        if self.access_token and time.time() < self.token_expires_at:
            return

        # Refresh token
        response = await self.http.post("/auth/token", data={
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
        })
        data = response.json()
        self.access_token = data["access_token"]
        self.token_expires_at = time.time() + data["expires_in"] - 60  # Buffer

    async def execute(self, command, args):
        await self.ensure_token()
        # Make request...

3. Use Correlation IDs for Debugging

import uuid

async def execute_command(client, command, args):
    correlation_id = str(uuid.uuid4())

    logger.info(f"Executing {command}", extra={
        "correlation_id": correlation_id,
        "args": args
    })

    result = await client.execute(command, args, headers={
        "X-Correlation-ID": correlation_id
    })

    logger.info(f"Command completed", extra={
        "correlation_id": correlation_id,
        "result": result
    })

    return result

4. Implement Circuit Breaker

from circuitbreaker import circuit

@circuit(failure_threshold=5, recovery_timeout=30)
async def execute_command(client, command, args):
    return await client.execute(command, args)