firsthandapi 0.1.0

Official Python SDK for FirstHandAPI — human-in-the-loop crowdsourced data collection API

firsthandapi — FirstHandAPI Python SDK

Official Python SDK for FirstHandAPI — the human-in-the-loop (HITL) crowdsourced data collection API.

Post jobs via API. Real human workers capture photos, audio, video, and screen recordings. AI ensemble (Claude Vision + Whisper) auto-scores quality. Approved files delivered to your S3 folder. Use cases: UGC collection, ground truth for AI evaluation, LLM training data.

Installation

pip install firsthandapi

Quick Start

from firsthandapi import FirstHandClient

client = FirstHandClient(api_key="fh_live_...")

# Submit a task for human review
task = client.create_task({
    "task_type": "support_reply",
    "category": "text",
    "input": {"customer_message": "I want a refund"},
    "candidate_output": {"content": "I can help with that..."},
})

print(task["id"])  # task_01JQ...

Async Usage

import asyncio
from firsthandapi import AsyncFirstHandClient

async def main():
    async with AsyncFirstHandClient(api_key="fh_live_...") as client:
        task = await client.create_task({
            "task_type": "support_reply",
            "category": "text",
            "input": {"customer_message": "I want a refund"},
            "candidate_output": {"content": "I can help with that..."},
        })

asyncio.run(main())

Submit and Wait

For simple workflows, submit a task and wait for the result in one call:

result = client.submit_and_wait(
    {
        "task_type": "support_reply",
        "category": "text",
        "input": {"customer_message": "I want a refund"},
        "candidate_output": {"content": "I can help with that..."},
    },
    max_wait_seconds=120,
)

if result["status"] == "resolved":
    print(result["decision"])  # approve, reject, approve_with_edit, escalate

Long Polling

Use the wait_seconds parameter to long-poll for task completion:

task = client.get_task("task_01JQ...", wait_seconds=30)

Webhook Signature Verification

from firsthandapi import verify_webhook_signature

try:
    verify_webhook_signature(
        payload=request.body,
        signature=request.headers["X-FirstHandAPI-Signature"],
        secret="whsec_...",
    )
    # Signature valid — process event
except ValueError as e:
    # Invalid signature
    print(f"Webhook verification failed: {e}")

Error Handling

from firsthandapi import FirstHandClient, FirstHandError

client = FirstHandClient(api_key="fh_live_...")

try:
    task = client.create_task({"task_type": "invalid"})
except FirstHandError as e:
    print(e.status)      # 400
    print(e.type)        # "validation_error"
    print(str(e))        # "Invalid task_type"
    print(e.request_id)  # "req_01JQ..."

Features

• Sync and async clients (FirstHandClient / AsyncFirstHandClient)
• Auto-retry with exponential backoff + jitter on 429 and 5xx errors
• Idempotency key generation on all POST requests
• Long-poll support via Prefer: wait=N header
• Webhook signature verification (HMAC-SHA256, timing-safe)
• Typed error hierarchy (FirstHandError, FirstHandRateLimitError, FirstHandConnectionError)

API Methods

Tasks

• create_task(body) — Submit a task for human review
• get_task(task_id) — Get task by ID (supports long-poll)
• list_tasks(**params) — List tasks with filters
• get_task_audit(task_id) — Get audit trail
• cancel_task(task_id) — Cancel a pending task
• validate_task(body) — Dry-run validation
• submit_and_wait(body) — Submit and poll until resolved

API Keys

• create_api_key(body) — Create a new API key
• list_api_keys() — List API keys
• rotate_api_key(key_id) — Rotate with 24h overlap
• revoke_api_key(key_id) — Revoke immediately

Webhooks

• create_webhook_endpoint(body) — Create endpoint
• list_webhook_endpoints() — List endpoints
• get_webhook_endpoint(id) — Get endpoint
• update_webhook_endpoint(id, body) — Update endpoint
• delete_webhook_endpoint(id) — Delete endpoint
• rotate_webhook_secret(id) — Rotate signing secret

Webhook Events

• list_webhook_events() — List events
• replay_webhook_event(id) — Replay an event

Billing

• get_credit_balance() — Current credit balance
• list_transactions() — Credit transaction history
• purchase_credits(body) — Initiate Stripe Checkout

Quality

• get_quality_summary() — Aggregate quality metrics

Organization

• get_settings() — Get org settings
• update_settings(body) — Update org settings

Requirements

• Python 3.9+
• httpx >= 0.27

License

MIT