Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.humancheck.dev/llms.txt

Use this file to discover all available pages before exploring further.

Humancheck uses an adapter pattern to support different AI frameworks. You can create custom adapters for frameworks that don’t have built-in support.

​
Adapter Pattern

Adapters convert framework-specific review requests into Humancheck’s universal UniversalReview format. This allows any framework to work with Humancheck.

​
Creating a Custom Adapter

​
Step 1: Create Adapter Class

from humancheck.adapters.base import BaseAdapter
from humancheck.schemas import UniversalReview, DecisionResponse

class MyFrameworkAdapter(BaseAdapter):
    """Adapter for MyFramework."""
    
    def get_adapter_name(self) -> str:
        """Return adapter identifier."""
        return "myframework"
    
    async def to_universal(self, framework_request: dict) -> UniversalReview:
        """Convert framework request to UniversalReview."""
        return UniversalReview(
            task_type=framework_request.get("action_type"),
            proposed_action=framework_request.get("action"),
            agent_reasoning=framework_request.get("reason"),
            confidence_score=framework_request.get("confidence"),
            urgency=framework_request.get("priority", "medium"),
            framework="myframework",
            metadata=framework_request.get("context", {}),
            blocking=framework_request.get("wait_for_decision", False)
        )
    
    async def from_universal(self, review: dict) -> dict:
        """Convert UniversalReview to framework format."""
        return {
            "review_id": review["id"],
            "status": review["status"],
            "action": review["proposed_action"],
            "decision": review.get("decision", {}).get("decision_type") if review.get("decision") else None
        }
    
    async def handle_blocking(self, review_id: int, timeout: float = 300.0) -> DecisionResponse:
        """Wait for decision and return it."""
        # Use the base implementation or customize
        return await super().handle_blocking(review_id, timeout)

​
Step 2: Register Adapter

from humancheck.adapters import register_adapter

# Register your adapter
adapter = MyFrameworkAdapter()
register_adapter(adapter)

​
Step 3: Use Adapter

from humancheck.adapters import get_adapter

# Get your adapter
adapter = get_adapter("myframework")

# Convert framework request to universal
framework_request = {
    "action_type": "payment",
    "action": "Process payment of $5,000",
    "reason": "High-value payment",
    "confidence": 0.85,
    "priority": "high",
    "wait_for_decision": True
}

universal_review = await adapter.to_universal(framework_request)

# Create review using API
async with httpx.AsyncClient() as client:
    response = await client.post(
        "https://api.humancheck.dev/reviews",
        json=universal_review.dict()
    )
    review = response.json()
    
    # If blocking, get decision
    if framework_request.get("wait_for_decision"):
        decision = await adapter.handle_blocking(review["id"])
        # Convert back to framework format
        framework_response = await adapter.from_universal(review)

​
Complete Example

Here’s a complete example for a hypothetical framework: 
"""Example adapter for MyFramework."""

import asyncio
import httpx
from typing import Optional
from humancheck.adapters.base import BaseAdapter
from humancheck.schemas import UniversalReview, DecisionResponse
from humancheck.database import get_db

class MyFrameworkAdapter(BaseAdapter):
    """Adapter for MyFramework."""
    
    def __init__(self, api_url: str = "https://api.humancheck.dev"):
        self.api_url = api_url
        self.db_session = get_db()
    
    def get_adapter_name(self) -> str:
        return "myframework"
    
    async def to_universal(self, framework_request: dict) -> UniversalReview:
        """Convert MyFramework request to UniversalReview."""
        # Map framework fields to universal format
        task_type = framework_request.get("operation_type")
        proposed_action = framework_request.get("command")
        reasoning = framework_request.get("explanation")
        confidence = framework_request.get("certainty", 1.0)
        
        # Map priority to urgency
        priority_map = {
            "low": "low",
            "normal": "medium",
            "high": "high",
            "urgent": "critical"
        }
        urgency = priority_map.get(
            framework_request.get("priority", "normal"),
            "medium"
        )
        
        return UniversalReview(
            task_type=task_type,
            proposed_action=proposed_action,
            agent_reasoning=reasoning,
            confidence_score=confidence,
            urgency=urgency,
            framework="myframework",
            metadata={
                "original_request": framework_request,
                "user_id": framework_request.get("user_id"),
                "session_id": framework_request.get("session_id")
            },
            blocking=framework_request.get("sync", False)
        )
    
    async def from_universal(self, review: dict) -> dict:
        """Convert UniversalReview to MyFramework format."""
        decision = review.get("decision")
        
        return {
            "review_id": review["id"],
            "status": review["status"],
            "command": review["proposed_action"],
            "approved": decision["decision_type"] == "approve" if decision else None,
            "modified_command": decision.get("modified_action") if decision and decision.get("decision_type") == "modify" else None,
            "rejection_reason": decision.get("notes") if decision and decision.get("decision_type") == "reject" else None
        }
    
    async def handle_blocking(self, review_id: int, timeout: float = 300.0) -> DecisionResponse:
        """Wait for decision with polling."""
        async with httpx.AsyncClient() as client:
            start_time = asyncio.get_event_loop().time()
            poll_interval = 2.0  # Poll every 2 seconds
            
            while True:
                elapsed = asyncio.get_event_loop().time() - start_time
                if elapsed > timeout:
                    raise TimeoutError(f"Decision timeout after {timeout}s")
                
                # Check for decision
                response = await client.get(
                    f"{self.api_url}/reviews/{review_id}/decision"
                )
                
                if response.status_code == 200:
                    decision_data = response.json()
                    if decision_data:
                        return DecisionResponse(**decision_data)
                
                await asyncio.sleep(poll_interval)

# Usage example
async def example_usage():
    """Example of using MyFramework adapter."""
    adapter = MyFrameworkAdapter()
    
    # Framework request
    framework_request = {
        "operation_type": "payment",
        "command": "transfer $5000 to account 12345",
        "explanation": "User requested transfer",
        "certainty": 0.8,
        "priority": "high",
        "sync": True,  # Wait for decision
        "user_id": 123,
        "session_id": "session-abc"
    }
    
    # Convert to universal
    universal_review = await adapter.to_universal(framework_request)
    
    # Create review
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.humancheck.dev/reviews",
            json=universal_review.dict()
        )
        review = response.json()
        review_id = review["id"]
        
        # Wait for decision (if blocking)
        if framework_request.get("sync"):
            decision = await adapter.handle_blocking(review_id)
            
            # Convert back to framework format
            framework_response = await adapter.from_universal(review)
            
            print(f"Decision: {framework_response}")

# Register adapter
from humancheck.adapters import register_adapter

adapter = MyFrameworkAdapter()
register_adapter(adapter)

​
Integration Patterns

​
Pattern 1: Direct Integration

# In your framework code
async def execute_with_review(action, context):
    adapter = get_adapter("myframework")
    review = await adapter.to_universal({
        "action": action,
        "context": context
    })
    
    # Create review via API
    review_obj = await create_review(review)
    
    # Wait for decision
    decision = await adapter.handle_blocking(review_obj.id)
    
    return decision

​
Pattern 2: Middleware Pattern

class HumancheckMiddleware:
    """Middleware for MyFramework."""
    
    def __init__(self):
        self.adapter = MyFrameworkAdapter()
    
    async def process(self, action, context):
        """Process action with human review."""
        # Check if review is needed
        if self.requires_review(action, context):
            review = await self.adapter.to_universal({
                "action": action,
                "context": context
            })
            
            review_obj = await create_review(review)
            decision = await self.adapter.handle_blocking(review_obj.id)
            
            if decision.decision_type == "approve":
                return await execute_action(action)
            elif decision.decision_type == "modify":
                return await execute_action(decision.modified_action)
            else:
                raise Exception("Action rejected")
        else:
            return await execute_action(action)

​
Pattern 3: Event-Based

class HumancheckEventHandler:
    """Event handler for framework events."""
    
    async def on_action_required(self, event):
        """Handle action requiring review."""
        adapter = get_adapter("myframework")
        review = await adapter.to_universal(event.data)
        
        # Create review
        review_obj = await create_review(review)
        
        # Emit event
        await emit("review_created", {
            "review_id": review_obj.id,
            "action": event.data["action"]
        })
    
    async def on_decision_received(self, event):
        """Handle decision received."""
        decision = event.data["decision"]
        
        # Emit event for framework to handle
        await emit("action_approved", {
            "action": decision.action,
            "decision": decision.decision_type
        })

​
Testing Your Adapter

import pytest
from humancheck.adapters import get_adapter

@pytest.mark.asyncio
async def test_adapter_conversion():
    """Test adapter conversion."""
    adapter = get_adapter("myframework")
    
    framework_request = {
        "action": "test action",
        "context": {"test": True}
    }
    
    universal = await adapter.to_universal(framework_request)
    assert universal.task_type is not None
    assert universal.proposed_action == "test action"
    
    # Convert back
    framework_response = await adapter.from_universal(universal.dict())
    assert framework_response["review_id"] is not None

@pytest.mark.asyncio
async def test_blocking_behavior():
    """Test blocking behavior."""
    adapter = get_adapter("myframework")
    
    # Create a test review
    review = await create_test_review()
    
    # Test timeout
    with pytest.raises(TimeoutError):
        await adapter.handle_blocking(review.id, timeout=1.0)

​
Best Practices

  1. Follow the adapter interface: Implement all required methods from BaseAdapter
  2. Handle errors gracefully: Implement proper error handling for network and timeout issues
  3. Provide clear mappings: Document how framework fields map to universal fields
  4. Support both blocking and non-blocking: Allow users to choose their preferred pattern
  5. Test thoroughly: Write tests for conversion, blocking behavior, and error cases

​
Next Steps