verifikio-sdk 1.2.0
Official Python SDK for Verifik.io trust infrastructure platform
Requires Python: >=3.7
Keywords: audit, logging, security, ai, agents, blockchain, verification, trust, infrastructure

# Verifik.io Python SDK

[![PyPI version](https://badge.fury.io/py/verifikio-sdk.svg)](https://badge.fury.io/py/verifikio-sdk)
[![Python Support](https://img.shields.io/pypi/pyversions/verifikio-sdk.svg)](https://pypi.org/project/verifikio-sdk/)
[![Downloads](https://img.shields.io/pypi/dm/verifikio-sdk)](https://pypi.org/project/verifikio-sdk/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)
[![](https://img.shields.io/badge/framework-CrewAI-blue)](https://github.com/joaomdmoura/crewAI)
[![](https://img.shields.io/badge/framework-LangChain-blue)](https://github.com/langchain-ai/langchain)

_The official Python SDK for Verifik.io – audit logs you can trust._

> **Also available:** [Node.js SDK](https://www.npmjs.com/package/@verifikio/sdk) - `npm install @verifikio/sdk`

📄 [Changelog](CHANGELOG.md) | 🚀 [v1.1.0 Release](CHANGELOG.md#110---2025-01-15)

## Installation

To install the official Verifik.io Python SDK:

```bash
pip install verifikio-sdk
```

For framework integrations:

```bash
# For CrewAI integration
pip install verifikio-sdk[crewai]

# For LangChain integration
pip install verifikio-sdk[langchain]

# For all integrations
pip install verifikio-sdk[all]
```

Verifik.io provides secure, immutable audit trails for AI agents using blockchain-style verification. Perfect for teams building with CrewAI, LangChain, AutoGPT, and other AI frameworks who need SOC 2 / ISO27001 compliance.

## Features

- 🔐 **Secure audit logging** with automatic hash chaining
- 🚀 **Developer-friendly** - Get started in minutes
- 📊 **Chain verification** - Ensure data integrity
- 🔑 **Simple authentication** with API keys
- ⚡ **Async support** - High-performance async operations with `async_mode=True`
- 📦 **Batch logging** - Log multiple events in a single API call
- 🐍 **Python 3.7+** compatible
- 📦 **Minimal dependencies** - Only requires `requests` (aiohttp for async)

## Quick Start

```python
from verifikio import VerifikClient

# Initialize client with your API key
client = VerifikClient(api_key="verifik_live_abc123...")

# Log an AI agent event
response = client.log_event(
    agent_name="data_processor",
    action="process_customer_data",
    inputs={"customer_id": "cust_123", "data_type": "profile"},
    outputs={"status": "processed", "record_count": 1},
    metadata={"processing_time": "1.2s"},
    workflow_id="workflow_456",
    status="success"
)

print(f"Audit log created: {response['id']}")
print(f"Hash: {response['hash']}")
```

## Framework Integrations

### CrewAI Integration

Automatically log all CrewAI agent actions and task executions:

```python
from verifikio import VerifikClient
from verifikio.integrations.crewai import VerifikCrewAIHandler
from crewai import Agent, Task, Crew

# Initialize Verifik.io
client = VerifikClient(api_key="verifik_live_xxx")
handler = VerifikCrewAIHandler(client, workflow_id="research-crew-001")

# Create agents with automatic logging
researcher = Agent(
    role="Researcher",
    goal="Research the latest AI trends",
    backstory="You are an expert AI researcher",
    callbacks=[handler]
)

writer = Agent(
    role="Writer", 
    goal="Write engaging content about AI",
    backstory="You are a technical writer",
    callbacks=[handler]
)

# Create tasks - all executions are logged
research_task = Task(
    description="Research GPT-4 capabilities",
    agent=researcher,
    callbacks=[handler]
)

write_task = Task(
    description="Write an article about the findings",
    agent=writer,
    callbacks=[handler]
)

# Run crew - all actions are tracked
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    callbacks=[handler]
)

results = crew.kickoff()
```

### LangChain Integration

Automatically log LangChain chains, tools, and LLM calls:

```python
from verifikio import VerifikClient
from verifikio.integrations.langchain import VerifikLangChainHandler
from langchain.chains import LLMChain
from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate

# Initialize Verifik.io
client = VerifikClient(api_key="verifik_live_xxx")
handler = VerifikLangChainHandler(
    client, 
    workflow_id="qa-chain-001",
    log_llm_calls=True  # Log individual LLM calls
)

# Create LLM with logging
llm = OpenAI(temperature=0.7, callbacks=[handler])

# Create chain with logging
prompt = PromptTemplate(
    input_variables=["topic"],
    template="Write a short article about {topic}"
)

chain = LLMChain(
    llm=llm, 
    prompt=prompt, 
    callbacks=[handler]
)

# All executions are automatically logged
result = chain.run("artificial intelligence in healthcare")
```

## Advanced Features

### ⚡ Async Support

For high-throughput workloads, Verifik.io SDK supports asynchronous logging using Python's asyncio:

```python
import asyncio
from verifikio import VerifikClient

async def main():
    # Initialize client with async_mode=True
    client = VerifikClient(api_key="verifik_live_xxx", async_mode=True)
    
    # All methods now return coroutines
    response = await client.log_event(
        agent_name="async-agent",
        action="process_data",
        inputs={"foo": "bar"},
        status="success"
    )
    print(f"Async log created: {response['id']}")
    
    # Clean up
    await client.close()

asyncio.run(main())
```

To enable async:
- Pass `async_mode=True` when creating the VerifikClient
- All client methods become await-able when async_mode is enabled
- Install with: `pip install verifikio-sdk[async]`

### 📦 Batch Logging

Batch mode allows efficient submission of multiple logs in a single API call:

```python
from verifikio import VerifikClient

client = VerifikClient(api_key="verifik_live_xxx")

# Prepare multiple events
batch = [
    {
        "agent_name": "agent-1",
        "action": "task-a",
        "inputs": {"data": 1},
        "status": "success"
    },
    {
        "agent_name": "agent-2",
        "action": "task-b",
        "inputs": {"data": 2},
        "status": "error"
    }
]

# Submit all at once
response = client.log_batch(batch)
print(f"Batch logged {len(response)} events")
```

Benefits:
- Reduce API calls for bulk auditing scenarios
- Minimize HTTP overhead
- Maintain data consistency across related events

### Combined Async + Batch Example

```python
async def process_large_dataset():
    client = VerifikClient(api_key="verifik_live_xxx", async_mode=True)
    
    # Process data in batches asynchronously
    batch = [
        {"agent_name": f"worker-{i}", "action": "process", "status": "success"}
        for i in range(100)
    ]
    
    response = await client.log_batch(batch)
    print(f"Logged {len(response)} events asynchronously")
    
    await client.close()
```

## API Reference

### VerifikClient

#### `__init__(api_key, base_url="https://api.verifik.io/v1", timeout=30, async_mode=False)`

Create a new Verifik.io client instance.

**Parameters:**
- `api_key` (str): Your Verifik.io API key (format: `verifik_live_*`)
- `base_url` (str, optional): Base URL for the API
- `timeout` (int, optional): Request timeout in seconds
- `async_mode` (bool, optional): Enable async operation mode (default: False)

```python
# Sync mode (default)
client = VerifikClient(api_key="verifik_live_abc123...")

# Async mode
client = VerifikClient(api_key="verifik_live_abc123...", async_mode=True)
```

#### `log_event(agent_name, action, **kwargs)`

Create a new audit log entry.

**Parameters:**
- `agent_name` (str): Name/identifier of the AI agent or service
- `action` (str): The action that was performed
- `inputs` (dict, optional): Input data/parameters for the action
- `outputs` (dict, optional): Output data/results from the action
- `metadata` (dict, optional): Additional metadata about the event
- `workflow_id` (str, optional): Workflow or session identifier
- `status` (str, optional): Status of the action (e.g., "success", "error", "pending")

**Returns:** `dict` - The created audit log entry

```python
response = client.log_event(
    agent_name="email_agent",
    action="send_notification",
    inputs={"recipient": "user@example.com"},
    outputs={"message_id": "msg_123"},
    status="success"
)
```

#### `get_logs(limit=50, offset=0, workflow_id=None)`

Retrieve audit logs with pagination.

**Parameters:**
- `limit` (int, optional): Number of logs to return (max 100)
- `offset` (int, optional): Number of logs to skip
- `workflow_id` (str, optional): Filter by workflow ID

**Returns:** `dict` - Paginated list of audit logs

```python
logs = client.get_logs(limit=20, offset=0)
for log in logs['logs']:
    print(f"Agent: {log['agentId']}, Action: {log['action']}")
```

#### `verify_chain()`

Verify the integrity of the audit log chain.

**Returns:** `dict` - Verification results

```python
verification = client.verify_chain()
print(f"Chain integrity: {verification['isValid']}")
```

#### `get_stats()`

Get account statistics and metrics.

**Returns:** `dict` - Account statistics

```python
stats = client.get_stats()
print(f"Total logs: {stats['totalLogs']}")
print(f"Chain integrity: {stats['chainIntegrity']}%")
```

#### `log_batch(events)`

Log multiple events in a single API request.

**Parameters:**
- `events` (list): List of event dictionaries, each containing:
  - `agent_name` (str, required): Name of the agent
  - `action` (str, required): Action performed
  - `inputs` (dict, optional): Input data
  - `outputs` (dict, optional): Output data
  - `metadata` (dict, optional): Additional metadata
  - `workflow_id` (str, optional): Workflow identifier
  - `status` (str, optional): Status of the action

**Returns:** `list` - List of created log entries

```python
batch = [
    {"agent_name": "agent1", "action": "process", "status": "success"},
    {"agent_name": "agent2", "action": "validate", "status": "success"}
]
response = client.log_batch(batch)
```

**Note:** In async mode (`async_mode=True`), all methods return coroutines and must be awaited.

#### `__init__(api_key, base_url="https://api.verifik.io/v1")`

Create a new async Verifik.io client instance.

**Parameters:**
- `api_key` (str): Your Verifik.io API key (format: `verifik_live_*`)
- `base_url` (str, optional): Base URL for the API

#### `log_event_async(agent_name, action, ...)`

Asynchronously log a single audit event. Same parameters as `log_event()`.

**Returns:** `dict` - The created audit log entry

#### `log_events_async(events)`

Asynchronously log multiple audit events in a single batch request.

**Parameters:**
- `events` (list): List of event dictionaries, each containing:
  - `agent_name` (str, required): Name of the agent
  - `action` (str, required): Action performed
  - All other parameters from `log_event()` are optional

**Returns:** `dict` - Response containing all created log entries

```python
events = [
    {"agent_name": "parser", "action": "parse_document", "status": "success"},
    {"agent_name": "validator", "action": "validate_data", "status": "success"},
    {"agent_name": "storage", "action": "save_results", "status": "success"}
]
response = await client.log_events_async(events)
```

#### `get_logs_async(limit=50, offset=0)`

Asynchronously retrieve audit logs. Same parameters and return as `get_logs()`.

#### `verify_chain_async()`

Asynchronously verify chain integrity. Same return as `verify_chain()`.

#### `get_stats_async()`

Asynchronously get account statistics. Same return as `get_stats()`.

#### `close()`

Close the underlying aiohttp session. Called automatically when using context manager.

## Framework Examples

### CrewAI Integration

```python
from verifikio import VerifikClient
from crewai import Agent, Task, Crew

# Initialize Verifik.io client
verifik = VerifikClient(api_key="verifik_live_abc123...")

# Create your CrewAI agents
researcher = Agent(
    role='Research Analyst',
    goal='Analyze market trends',
    backstory='Expert in market analysis'
)

# Custom callback to log CrewAI events
def log_crew_event(agent_name, action, inputs=None, outputs=None, status="success"):
    verifik.log_event(
        agent_name=agent_name,
        action=action,
        inputs=inputs,
        outputs=outputs,
        metadata={"framework": "crewai"},
        status=status
    )

# Log when agent starts task
log_crew_event("research_analyst", "start_analysis", {"topic": "AI market"})

# ... run your CrewAI workflow ...

# Log completion
log_crew_event("research_analyst", "complete_analysis", 
              outputs={"findings": "Market growing 40% YoY"})
```

### LangChain Integration

```python
from verifikio import VerifikClient
from langchain.chains import LLMChain
from langchain.callbacks.base import BaseCallbackHandler

verifik = VerifikClient(api_key="verifik_live_abc123...")

class VerifikCallback(BaseCallbackHandler):
    def on_chain_start(self, serialized, inputs, **kwargs):
        verifik.log_event(
            agent_name="langchain_agent",
            action="chain_start",
            inputs=inputs,
            metadata={"chain_type": serialized.get("name", "unknown")}
        )
    
    def on_chain_end(self, outputs, **kwargs):
        verifik.log_event(
            agent_name="langchain_agent",
            action="chain_end",
            outputs=outputs,
            status="success"
        )

# Use the callback in your LangChain
chain = LLMChain(llm=llm, prompt=prompt, callbacks=[VerifikCallback()])
```

## Error Handling

The SDK includes comprehensive error handling:

```python
from verifikio import VerifikClient, AuthenticationError, ValidationError, APIError

client = VerifikClient(api_key="verifik_live_abc123...")

try:
    response = client.log_event(
        agent_name="test_agent",
        action="test_action"
    )
except AuthenticationError:
    print("Invalid API key")
except ValidationError as e:
    print(f"Invalid parameters: {e}")
except APIError as e:
    print(f"API error: {e}")
```

## Context Manager Support

The client supports context manager protocol for automatic cleanup:

```python
with VerifikClient(api_key="verifik_live_abc123...") as client:
    client.log_event(
        agent_name="context_agent",
        action="test_action"
    )
# Client session automatically closed
```

## Getting Your API Key

1. Sign up at [verifik.io](https://verifik.io)
2. Go to your Settings page
3. Generate a new API key
4. Copy the key (format: `verifik_live_...`)

## Development

### Running Tests

```bash
# Install development dependencies
pip install -e ".[dev]"

# Run tests
pytest tests/

# Run tests with coverage
pytest tests/ --cov=verifikio
```

### Code Quality

```bash
# Format code
black verifikio/

# Lint code
flake8 verifikio/

# Type checking
mypy verifikio/
```

## Contributing

1. Fork the repository
2. Create a feature branch: `git checkout -b feature-name`
3. Make your changes and add tests
4. Run the test suite: `pytest`
5. Submit a pull request

## License

This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.

## Support

- 📧 **Email**: support@verifik.io
- 📖 **Documentation**: https://docs.verifik.io
- 🐛 **Bug Reports**: https://github.com/verifik-io/verifikio-python/issues
- 💬 **Community**: https://discord.gg/verifik

## Changelog

### 1.0.0 (2024-01-14)
- Initial release
- Core audit logging functionality
- Chain verification
- Full API coverage
- Python 3.7+ support