ai-traceability-sdk 1.0.0

A comprehensive Python SDK for AI activity tracking and monitoring

AI Traceability Python SDK

A comprehensive Python SDK for integrating with the AI Traceability Service. Provides async/sync clients, session management, batching, and comprehensive tool call tracking following the MCP protocol.

Features

• Async/Sync Support: Full async and sync client implementations
• Session Management: Automatic session lifecycle management
• Batch Processing: Efficient batching for high-throughput scenarios
• MCP Protocol: Complete Model Context Protocol support for tool calls
• Type Safety: Full type hints and Pydantic model validation
• Error Handling: Comprehensive error handling with custom exceptions
• Authentication: JWT and API key authentication support
• Production Ready: Enterprise-grade reliability and security

Installation

From PyPI (when published)

pip install ai-traceability-sdk

Local Development Install

# Clone the repository
git clone <repository-url>
cd python-sdk

# Install in development mode
pip install -e .

# Or install with development dependencies
pip install -e .[dev]

From Source

# Install from the python-sdk directory
pip install ./python-sdk

# Or install with all dependencies
pip install ./python-sdk[dev,docs]

Quick Start

Basic Usage

import asyncio
from python_sdk import AITraceabilityClient
from python_sdk.models import ActivityData

async def main():
    # Initialize the client
    client = AITraceabilityClient(
        base_url="https://your-api-endpoint.com",
        api_key="your-api-key"
    )
    
    # Track an activity
    activity = ActivityData(
        agent_id="my-agent",
        activity_type="chat",
        content="User asked about weather"
    )
    
    response = await client.track_activity(activity)
    print(f"Activity tracked: {response.id}")

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

Session Management

from python_sdk import AITraceabilityClient, Session

async def session_example():
    client = AITraceabilityClient(
        base_url="https://your-api-endpoint.com",
        api_key="your-api-key"
    )
    
    # Start a session
    session = await client.start_session(
        agent_id="my-agent",
        user_id="user123"
    )
    
    # Track activities within the session
    await session.track_activity(
        activity_type="chat",
        content="Hello, how can I help you?"
    )
    
    # End the session
    await session.end()

Batch Processing

from python_sdk import AITraceabilityClient, BatchProcessor
from python_sdk.models import ActivityData

async def batch_example():
    client = AITraceabilityClient(
        base_url="https://your-api-endpoint.com", 
        api_key="your-api-key"
    )
    
    # Create batch processor
    batch = BatchProcessor(client, batch_size=100)
    
    # Add multiple activities
    activities = [
        ActivityData(agent_id="agent1", activity_type="chat", content="Message 1"),
        ActivityData(agent_id="agent1", activity_type="chat", content="Message 2"),
        ActivityData(agent_id="agent1", activity_type="chat", content="Message 3"),
    ]
    
    for activity in activities:
        await batch.add_activity(activity)
    
    # Flush remaining items
    await batch.flush()

Tool Call Tracking

from python_sdk.models import ToolCallData

async def tool_call_example():
    client = AITraceabilityClient(
        base_url="https://your-api-endpoint.com",
        api_key="your-api-key"
    )
    
    # Track a tool call
    tool_call = ToolCallData(
        agent_id="my-agent",
        session_id="session-123",
        tool_name="weather_api",
        tool_arguments={"location": "New York"},
        tool_result={"temperature": "72°F", "conditions": "sunny"}
    )
    
    response = await client.track_tool_call(tool_call)
    print(f"Tool call tracked: {response.id}")

Configuration

Environment Variables

# API Configuration
TRACEABILITY_API_KEY=your-api-key
TRACEABILITY_URL=https://your-api-endpoint.com

# Optional Configuration
TRACEABILITY_TIMEOUT=30
TRACEABILITY_MAX_RETRIES=3
TRACEABILITY_BATCH_SIZE=100

Programmatic Configuration

from python_sdk import AITraceabilityClient
from python_sdk.config import SDKConfig

# Custom configuration
config = SDKConfig(
    api_key="your-api-key",
    base_url="https://your-api-endpoint.com",
    timeout=30,
    max_retries=3,
    batch_size=100
)

client = AITraceabilityClient(config=config)

Data Models

ActivityData

from python_sdk.models import ActivityData

activity = ActivityData(
    agent_id="my-agent",
    activity_type="chat",
    content="User interaction",
    user_id="user123",           # Optional
    session_id="session-456",    # Optional
    metadata={"key": "value"}    # Optional
)

ToolCallData

from python_sdk.models import ToolCallData

tool_call = ToolCallData(
    agent_id="my-agent",
    session_id="session-123",
    tool_name="api_call",
    tool_arguments={"param": "value"},
    tool_result={"success": True},
    duration_ms=1500             # Optional
)

SessionData

from python_sdk.models import SessionData

session = SessionData(
    agent_id="my-agent",
    user_id="user123",
    session_type="chat",         # Optional
    metadata={"context": "support"}  # Optional
)

Error Handling

from python_sdk.exceptions import (
    TraceabilityError,
    AuthenticationError,
    RateLimitError,
    ValidationError,
    NetworkError
)

try:
    response = await client.track_activity(activity)
except AuthenticationError:
    print("Invalid API key")
except RateLimitError as e:
    print(f"Rate limited. Retry after: {e.retry_after} seconds")
except ValidationError as e:
    print(f"Invalid data: {e.message}")
except NetworkError:
    print("Network connectivity issue")
except TraceabilityError as e:
    print(f"General error: {e.message}")

Development

Setup Development Environment

# Clone repository
git clone <repository-url>
cd python-sdk

# Install with development dependencies
pip install -e .[dev]

# Run tests
pytest

# Format code
black .
isort .

# Type checking
mypy python_sdk

Running Tests

# Run all tests
pytest

# Run with coverage
pytest --cov=python_sdk --cov-report=html

# Run specific test types
pytest -m unit        # Unit tests only
pytest -m integration # Integration tests only

API Reference

Client Methods

• track_activity(activity: ActivityData) -> ActivityResponse
• track_tool_call(tool_call: ToolCallData) -> ToolCallResponse
• track_decision(decision: DecisionData) -> DecisionResponse
• start_session(agent_id: str, user_id: str = None, **kwargs) -> Session
• batch_activities(activities: List[ActivityData]) -> List[ActivityResponse]

Session Methods

• track_activity(activity_type: str, content: str, **kwargs) -> ActivityResponse
• track_tool_call(tool_name: str, arguments: dict, result: dict, **kwargs) -> ToolCallResponse
• end() -> None

License

This project is licensed under the MIT License - see the LICENSE file for details.

Contributing

1. Fork the repository
2. Create a feature branch (git checkout -b feature/amazing-feature)
3. Commit your changes (git commit -m 'Add some amazing feature')
4. Push to the branch (git push origin feature/amazing-feature)
5. Open a Pull Request

Support

• Documentation: https://docs.ai-traceability.com/
• Issues: GitHub Issues
• Email: [e-mail]