tupl-sdk 1.2.1

Python SDK for Semantic Security - capture and send IntentEvents with LangGraph integration

Tupl Python SDK

Python client library for the Semantic Security MVP - capture and send IntentEvents to the Management Plane for policy enforcement.

Installation

# Install with uv (recommended)
uv pip install -e .

# Or with pip
pip install -e .

Quick Start

from tupl import TuplClient, IntentEvent, Actor, Resource, Data, Risk
import time
import uuid

# Create client
client = TuplClient(endpoint="http://localhost:8000")

# Create an intent event
event = IntentEvent(
    id=f"evt-{uuid.uuid4()}",
    tenantId="tenant-123",
    timestamp=time.time(),
    actor=Actor(id="user-alice", type="user"),
    action="read",
    resource=Resource(type="database", name="users_db", location="cloud"),
    data=Data(categories=["pii"], pii=True, volume="row"),
    risk=Risk(authn="mfa", network="corp", timeOfDay=14)
)

# Send to Management Plane
result = client.capture(event)

if result:
    print(f"Decision: {'ALLOW' if result.decision == 1 else 'BLOCK'}")
    print(f"Similarities: {result.slice_similarities}")

client.close()

Features

Immediate Mode (Default)

Send events immediately and get synchronous responses:

client = TuplClient(endpoint="http://localhost:8000")
result = client.capture(event)  # Blocks until response

Buffered Mode

Buffer events and send in batches for better performance:

client = TuplClient(
    endpoint="http://localhost:8000",
    buffered=True,
    buffer_size=10,      # Flush after 10 events
    buffer_timeout=5.0   # Or flush after 5 seconds
)

client.capture(event1)  # Returns None (buffered)
client.capture(event2)  # Returns None (buffered)
# ... automatically flushes when buffer is full or timeout occurs

client.flush()  # Manual flush
client.close()  # Flushes remaining events

Async Support

from tupl import AsyncTuplClient

async with AsyncTuplClient(endpoint="http://localhost:8000") as client:
    result = await client.capture(event)

Context Manager

with TuplClient(endpoint="http://localhost:8000") as client:
    result = client.capture(event)
# Automatically closes and flushes

Configuration

TuplClient Options

• endpoint (str): Management Plane base URL (default: http://localhost:8000)
• api_version (str): API version (default: v1)
• buffered (bool): Enable event buffering (default: False)
• buffer_size (int): Max events before auto-flush (default: 10)
• buffer_timeout (float): Seconds before auto-flush (default: 5.0)
• timeout (float): HTTP request timeout in seconds (default: 10.0)
• retry_count (int): Number of retries on failure (default: 3)

Data Types

IntentEvent

Structured record of an LLM/tool call intent:

• id: Unique event ID (UUID)
• schemaVersion: Schema version (always "v1")
• tenantId: Tenant identifier
• timestamp: Unix timestamp
• actor: Who initiated the action
• action: What action ("read", "write", "delete", "export", "execute", "update")
• resource: What resource is being accessed
• data: Data characteristics
• risk: Risk context

Actor

• id: Actor identifier
• type: "user" or "service"

Resource

• type: Resource type ("database", "file", "api", "service", "user_data")
• name: Optional resource name
• location: Optional location ("local", "cloud", "external")

Data

• categories: List of data categories ("pii", "financial", "medical", "public", "internal")
• pii: Optional boolean indicating PII data
• volume: Optional volume class ("row", "table", "dump", "bulk")

Risk

• authn: Authentication level ("none", "user", "mfa", "service")
• network: Network context ("corp", "vpn", "public")
• timeOfDay: Optional hour of day (0-23)

ComparisonResult

Response from Management Plane:

• decision: 0 = block, 1 = allow
• slice_similarities: List of 4 floats (action, resource, data, risk similarity scores)

Examples

See examples/basic_usage.py for a complete example.

# Run basic example
cd tupl_sdk/python
uv run python examples/basic_usage.py

Testing

# Run unit tests
uv run pytest tests/test_client.py -v

# Run with coverage
uv run pytest tests/test_client.py --cov=tupl --cov-report=html

Development

Project Structure

tupl_sdk/python/
├── tupl/                  # Main package
│   ├── __init__.py       # Public API exports
│   ├── types.py          # Pydantic data models
│   ├── client.py         # TuplClient + AsyncTuplClient
│   └── buffer.py         # EventBuffer for batching
├── tests/                # Unit tests
│   ├── __init__.py
│   └── test_client.py
├── examples/             # Usage examples
│   ├── __init__.py
│   └── basic_usage.py
├── pyproject.toml        # Package configuration
└── README.md             # This file

Dependencies

• Python 3.14+
• httpx (HTTP client)
• pydantic (data validation)

Code Style

• Type hints on all functions
• Pydantic models for all data structures
• No Dict[str, Any] fields (Google GenAI compatibility)
• Comprehensive docstrings

Troubleshooting

Connection Errors

Ensure the Management Plane is running:

cd management-plane
./run.sh
# Server should start on http://localhost:8000

Test health endpoint:

curl http://localhost:8000/health

Import Errors

Install the package in development mode:

uv pip install -e .

License

See root project LICENSE file.

Support

Report issues at: https://github.com/your-org/mgmt-plane/issues