Python SDK Reference

pip install clawauth-sdk

# Or with development dependencies
pip install "clawauth-sdk[dev]"

from clawauth import ClawAuth

auth = ClawAuth(
    service_id="your-service-id",
    api_key="your-api-key",
    base_url="https://api.clawauth.com"  # optional
)

# Synchronous
result = auth.verify(agent_token)

if result.valid:
    print(f"Agent ID: {result.agent.id}")
    print(f"Agent name: {result.agent.name}")
    print(f"Trust score: {result.agent.trust_score}")
    print(f"Roles: {result.agent.roles}")
    print(f"Permissions: {result.agent.scopes}")
else:
    print(f"Invalid token: {result.error}")

# Asynchronous
import asyncio

async def verify_agent():
    result = await auth.verify_async(agent_token)
    return result

result = asyncio.run(verify_agent())

from pydantic import BaseModel
from typing import Optional, List

class Agent(BaseModel):
    id: str
    name: str
    email: str
    trust_score: float
    roles: List[str]
    scopes: List[str]

class VerifyTokenResponse(BaseModel):
    valid: bool
    agent: Optional[Agent] = None
    error: Optional[str] = None

# Synchronous
agent = auth.get_agent("agent_abc123")

print(f"Agent details: {agent.name}")
print(f"Trust score: {agent.trust_score}")
print(f"Status: {agent.status}")

# Asynchronous
agent = await auth.get_agent_async("agent_abc123")

# Synchronous
auth.assign_role("agent_abc123", "admin")
auth.assign_role("agent_def456", "user")

# Asynchronous
await auth.assign_role_async("agent_abc123", "admin")

# Synchronous
auth.revoke_role("agent_abc123", "admin")

# Asynchronous
await auth.revoke_role_async("agent_abc123", "admin")

# Positive feedback
auth.report_trust(
    "agent_abc123", 
    score=9.0, 
    reason="Excellent API usage, no rate limit violations"
)

# Negative feedback
auth.report_trust(
    "agent_def456", 
    score=3.0, 
    reason="Frequent authentication failures"
)

# Asynchronous
await auth.report_trust_async("agent_abc123", score=8.5)

from clawauth import ClawAuthAgent

# Using private key file
agent = ClawAuthAgent(
    agent_id="your-agent-id",
    private_key_path="/path/to/key.pem",
    base_url="https://api.clawauth.com"  # optional
)

# Using private key bytes
with open("key.pem", "rb") as f:
    private_key_data = f.read()

agent = ClawAuthAgent(
    agent_id="your-agent-id",
    private_key=private_key_data,
    base_url="https://api.clawauth.com"
)

keypair = ClawAuthAgent.generate_keypair()

print("Public key (share with ClawAuth):")
print(keypair["public_key"])

print("Private key (keep secret):")
print(keypair["private_key"])

# Save private key securely
with open("agent-key.pem", "w") as f:
    f.write(keypair["private_key"])
    
# Set secure permissions
import os
os.chmod("agent-key.pem", 0o600)

# Synchronous
registration = ClawAuthAgent.register(
    email="[email protected]",
    name="My AI Agent",
    public_key=keypair["public_key"],
    base_url="https://api.clawauth.com"  # optional
)

print(f"Agent registered with ID: {registration.agent_id}")
print(f"Status: {registration.status}")

# Asynchronous
registration = await ClawAuthAgent.register_async(
    email="[email protected]",
    name="My AI Agent",
    public_key=keypair["public_key"]
)

# After receiving email with verification code
result = agent.verify_email(
    registration.registration_id,
    "123456",  # code from email
    "signature_from_email"  # cryptographic proof
)

# Asynchronous
result = await agent.verify_email_async(
    registration.registration_id, 
    "123456", 
    "signature_from_email"
)

# Synchronous
tokens = agent.authenticate()

print(f"Access token: {tokens['access_token']}")
print(f"Refresh token: {tokens['refresh_token']}")
print(f"Expires in: {tokens['expires_in']} seconds")

# Asynchronous
tokens = await agent.authenticate_async()

# Use access token with services
import httpx

async with httpx.AsyncClient() as client:
    response = await client.get(
        "https://api.example.com/data",
        headers={
            "Authorization": f"Bearer {tokens['access_token']}"
        }
    )

# Synchronous
new_tokens = agent.refresh(tokens["refresh_token"])

print(f"New access token: {new_tokens['access_token']}")
print(f"New refresh token: {new_tokens['refresh_token']}")
print(f"Expires in: {new_tokens['expires_in']} seconds")

# Asynchronous
new_tokens = await agent.refresh_async(tokens["refresh_token"])

from clawauth.exceptions import (
    ClawAuthError,
    AuthenticationError,
    InvalidSignatureError,
    InsufficientPermissionsError,
    RateLimitError
)

try:
    result = auth.verify(token)
except AuthenticationError as e:
    print(f"Authentication failed: {e.message}")
except RateLimitError as e:
    print("Rate limit exceeded, try again later")
except ClawAuthError as e:
    print(f"ClawAuth error: {e.code} - {e.message}")
except Exception as e:
    print(f"Unknown error: {e}")

from clawauth import ClawAuth
from clawauth.types import VerifyTokenResponse, Agent

auth = ClawAuth(service_id="test-service", api_key="test-key")

# Type-safe verification
result: VerifyTokenResponse = auth.verify(token)

if result.valid and result.agent:
    # result.agent is properly typed
    agent: Agent = result.agent
    print(f"Trust score: {agent.trust_score}")

from fastapi import FastAPI, HTTPException, Depends
from fastapi.security import HTTPBearer
import asyncio
from clawauth import ClawAuth
from clawauth.types import Agent

app = FastAPI()
auth = ClawAuth(
    service_id=os.getenv("CLAWAUTH_SERVICE_ID"),
    api_key=os.getenv("CLAWAUTH_API_KEY")
)

security = HTTPBearer()

async def get_current_agent(token: str = Depends(security)) -> Agent:
    """Dependency to get authenticated agent."""
    result = await auth.verify_async(token.credentials)
    
    if not result.valid or not result.agent:
        raise HTTPException(status_code=401, detail="Invalid token")
    
    return result.agent

@app.get("/public")
async def public_endpoint():
    return {"message": "Hello World"}

@app.get("/protected")
async def protected_endpoint(agent: Agent = Depends(get_current_agent)):
    if agent.trust_score < 5.0:
        raise HTTPException(status_code=403, detail="Trust score too low")
    
    # Provide feedback
    await auth.report_trust_async(
        agent.id, 
        score=8.0, 
        reason="Successful API call"
    )
    
    return {
        "message": f"Hello {agent.name}!",
        "trust_score": agent.trust_score
    }

import asyncio
from django.http import JsonResponse
from django.utils.deprecation import MiddlewareMixin
from clawauth import ClawAuth

class ClawAuthMiddleware(MiddlewareMixin):
    def __init__(self, get_response):
        super().__init__(get_response)
        self.auth = ClawAuth(
            service_id=settings.CLAWAUTH_SERVICE_ID,
            api_key=settings.CLAWAUTH_API_KEY
        )
    
    def process_request(self, request):
        auth_header = request.META.get('HTTP_AUTHORIZATION', '')
        
        if auth_header.startswith('Bearer '):
            token = auth_header[7:]
            
            try:
                # Run async verification in sync context
                result = asyncio.run(self.auth.verify_async(token))
                
                if result.valid and result.agent:
                    request.agent = result.agent
                else:
                    request.agent = None
            except Exception:
                request.agent = None
        else:
            request.agent = None

# In views.py
from django.http import JsonResponse
from django.views.decorators.http import require_http_methods

@require_http_methods(["GET"])
def protected_view(request):
    if not hasattr(request, 'agent') or request.agent is None:
        return JsonResponse(
            {"error": "Authentication required"}, 
            status=401
        )
    
    return JsonResponse({
        "message": f"Hello {request.agent.name}!",
        "trust_score": request.agent.trust_score
    })

import asyncio
import os
from fastapi import FastAPI, HTTPException, Depends
from fastapi.security import HTTPBearer
from clawauth import ClawAuth
from clawauth.types import Agent

app = FastAPI(title="ClawAuth Demo Service")

# Initialize ClawAuth
auth = ClawAuth(
    service_id=os.getenv("CLAWAUTH_SERVICE_ID"),
    api_key=os.getenv("CLAWAUTH_API_KEY")
)

security = HTTPBearer()

async def authenticate_agent(token: str = Depends(security)) -> Agent:
    """Authenticate agent and return agent info."""
    result = await auth.verify_async(token.credentials)
    
    if not result.valid or not result.agent:
        raise HTTPException(
            status_code=401, 
            detail=result.error or "Invalid token"
        )
    
    return result.agent

@app.get("/")
async def root():
    return {"message": "ClawAuth Demo Service"}

@app.get("/profile")
async def get_profile(agent: Agent = Depends(authenticate_agent)):
    """Get authenticated agent's profile."""
    return {
        "id": agent.id,
        "name": agent.name,
        "trust_score": agent.trust_score,
        "roles": agent.roles
    }

@app.post("/trust-feedback")
async def submit_trust_feedback(
    target_agent_id: str,
    score: float,
    reason: str,
    agent: Agent = Depends(authenticate_agent)
):
    """Submit trust feedback (only for admins)."""
    if "admin" not in agent.roles:
        raise HTTPException(status_code=403, detail="Admin role required")
    
    if not 0 <= score <= 10:
        raise HTTPException(status_code=400, detail="Score must be between 0 and 10")
    
    await auth.report_trust_async(target_agent_id, score, reason)
    
    return {"success": True, "message": "Trust feedback submitted"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

import asyncio
import time
from pathlib import Path
from clawauth import ClawAuthAgent
import httpx

class MyAgent:
    def __init__(self, agent_id: str, private_key_path: str):
        self.auth = ClawAuthAgent(
            agent_id=agent_id,
            private_key_path=private_key_path
        )
        self.access_token = None
        self.refresh_token = None
        self.token_expires_at = 0
    
    async def initialize(self):
        """Initialize agent and authenticate."""
        tokens = await self.auth.authenticate_async()
        self.access_token = tokens["access_token"]
        self.refresh_token = tokens["refresh_token"]
        self.token_expires_at = time.time() + tokens["expires_in"]
        print("Agent authenticated successfully")
    
    async def _ensure_valid_token(self):
        """Ensure we have a valid access token."""
        if time.time() >= self.token_expires_at - 60:  # Refresh 1 min before expiry
            print("Refreshing access token...")
            tokens = await self.auth.refresh_async(self.refresh_token)
            self.access_token = tokens["access_token"]
            self.refresh_token = tokens["refresh_token"]
            self.token_expires_at = time.time() + tokens["expires_in"]
    
    async def make_authenticated_request(
        self, 
        url: str, 
        method: str = "GET", 
        **kwargs
    ) -> httpx.Response:
        """Make an authenticated HTTP request."""
        await self._ensure_valid_token()
        
        headers = kwargs.get("headers", {})
        headers["Authorization"] = f"Bearer {self.access_token}"
        kwargs["headers"] = headers
        
        async with httpx.AsyncClient() as client:
            response = await client.request(method, url, **kwargs)
            return response
    
    async def get_data(self):
        """Example API call."""
        response = await self.make_authenticated_request(
            "https://api.example.com/data"
        )
        return response.json()

# Usage example
async def main():
    # Create agent
    agent = MyAgent("my-agent-id", "agent-key.pem")
    
    # Initialize and authenticate
    await agent.initialize()
    
    # Make authenticated requests
    try:
        data = await agent.get_data()
        print("Data received:", data)
    except Exception as e:
        print(f"Error: {e}")

if __name__ == "__main__":
    asyncio.run(main())

import pytest
import asyncio
from clawauth import ClawAuth, ClawAuthAgent

@pytest.mark.asyncio
async def test_agent_authentication():
    """Test agent authentication flow."""
    # Generate test keypair
    keypair = ClawAuthAgent.generate_keypair()
    
    # This would normally be done once during setup
    registration = await ClawAuthAgent.register_async(
        email="[email protected]",
        name="Test Agent",
        public_key=keypair["public_key"]
    )
    
    # Create agent with generated key
    agent = ClawAuthAgent(
        agent_id=registration.agent_id,
        private_key=keypair["private_key"].encode()
    )
    
    # Test authentication
    tokens = await agent.authenticate_async()
    
    assert "access_token" in tokens
    assert "refresh_token" in tokens
    assert tokens["expires_in"] > 0

@pytest.mark.asyncio
async def test_service_verification():
    """Test service token verification."""
    auth = ClawAuth(
        service_id="test-service",
        api_key="test-key"
    )
    
    # This would be a real token in practice
    fake_token = "eyJhbGciOiJFUzI1Ni..."
    
    result = await auth.verify_async(fake_token)
    
    # Should fail with fake token
    assert not result.valid
    assert result.error is not None