Custom Adapters

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

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

Next Steps

Review the base adapter implementation
Check out existing adapters for reference
Submit your adapter as a pull request to help others!

Getting started

Products

Integrations

Use cases

Configuration

Platform Features

Connectors

Dashboard

Adapter Pattern

Creating a Custom Adapter

Step 1: Create Adapter Class

Step 2: Register Adapter

Step 3: Use Adapter

Complete Example

Integration Patterns

Pattern 1: Direct Integration

Pattern 2: Middleware Pattern

Pattern 3: Event-Based

Testing Your Adapter

Best Practices

Next Steps

​Adapter Pattern

​Creating a Custom Adapter

​Step 1: Create Adapter Class

​Step 2: Register Adapter

​Step 3: Use Adapter

​Complete Example

​Integration Patterns

​Pattern 1: Direct Integration

​Pattern 2: Middleware Pattern

​Pattern 3: Event-Based

​Testing Your Adapter

​Best Practices

​Next Steps

Adapter Pattern

Creating a Custom Adapter

Step 1: Create Adapter Class

Step 2: Register Adapter

Step 3: Use Adapter

Complete Example

Integration Patterns

Pattern 1: Direct Integration

Pattern 2: Middleware Pattern

Pattern 3: Event-Based

Testing Your Adapter

Best Practices

Next Steps