fulcrum-governance 0.1.4
Python SDK for Fulcrum - Intelligent AI Governance Platform
Requires Python: >=3.9
Keywords: ai, governance, llm, agents, safety, compliance

# Fulcrum Python SDK

> Intelligent AI Governance for Enterprise Agents

[![PyPI version](https://badge.fury.io/py/fulcrum-governance.svg)](https://badge.fury.io/py/fulcrum-governance)
[![Python 3.9+](https://img.shields.io/badge/python-3.9+-blue.svg)](https://www.python.org/downloads/)
[![License](https://img.shields.io/badge/License-Proprietary-blue.svg)](https://fulcrum.dev/license)

## Installation

```bash
pip install fulcrum-governance
```

## Quick Start

```python
from fulcrum import FulcrumClient

# Initialize client
client = FulcrumClient(
    host="your-fulcrum-server:50051",
    api_key="your-api-key"
)

# Wrap agent executions in governance envelopes
with client.envelope(workflow_id="customer-support-bot") as env:
    # Check if action is allowed before executing
    if env.guard("send_email", input_text=user_message):
        # Action approved - proceed
        result = send_email(user_message)
        env.log("email_sent", {"recipient": email, "status": "success"})
    else:
        # Action blocked by policy
        env.log("action_blocked", {"reason": "policy_violation"})
```

## Features

- **Policy Enforcement**: Real-time governance checks before agent actions
- **Cost Tracking**: Monitor and control LLM spending per workflow
- **Audit Trail**: Complete execution history for compliance
- **Fail-Safe Modes**: Configurable FAIL_OPEN or FAIL_CLOSED behavior

## Configuration

### Client Options

```python
from fulcrum import FulcrumClient, FailureMode

client = FulcrumClient(
    host="localhost:50051",          # Fulcrum server address
    api_key="your-api-key",          # API key for authentication
    tenant_id="your-tenant-id",      # Default tenant ID
    on_failure=FailureMode.FAIL_OPEN,  # FAIL_OPEN or FAIL_CLOSED
    timeout_ms=500,                  # Request timeout in milliseconds
    enable_tls=True,                 # Enable TLS encryption
    ca_cert_path="/path/to/ca.crt",  # Custom CA certificate (optional)
)
```

### Environment Variables

```bash
export FULCRUM_HOST="localhost:50051"
export FULCRUM_API_KEY="your-api-key"
export FULCRUM_TENANT_ID="your-tenant-id"
export FULCRUM_TIMEOUT_MS="500"
```

```python
# Client auto-discovers from environment
client = FulcrumClient.from_env()
```

## API Reference

### FulcrumClient

The main client for interacting with Fulcrum.

```python
client = FulcrumClient(host, api_key, **options)
```

#### Methods

| Method | Description |
|--------|-------------|
| `envelope(workflow_id, **kwargs)` | Create a governance envelope |
| `evaluate(action, input_text, **context)` | Evaluate a policy decision |
| `get_cost(envelope_id)` | Get cost for an envelope |
| `list_policies(tenant_id)` | List active policies |
| `health_check()` | Check server connectivity |

### Envelope

Context manager for governed executions.

```python
with client.envelope(
    workflow_id="my-workflow",
    execution_id="optional-custom-id",  # Auto-generated if not provided
    metadata={"user": "alice"},
) as env:
    # Governed execution
    pass
```

#### Envelope Methods

| Method | Description |
|--------|-------------|
| `guard(action, input_text, **metadata)` | Check if action is allowed |
| `log(event_type, payload)` | Log an event for audit |
| `checkpoint()` | Create execution checkpoint |
| `get_cost()` | Get current execution cost |

## Error Handling

```python
from fulcrum import FulcrumClient
from fulcrum.exceptions import (
    FulcrumError,           # Base exception
    PolicyViolationError,   # Action blocked by policy
    BudgetExceededError,    # Budget limit reached
    ConnectionError,        # Server unreachable
    AuthenticationError,    # Invalid API key
    TimeoutError,           # Request timed out
)

client = FulcrumClient(host="localhost:50051", api_key="key")

try:
    with client.envelope(workflow_id="my-agent") as env:
        if env.guard("send_email", input_text="Hello"):
            send_email("Hello")
except PolicyViolationError as e:
    print(f"Policy violation: {e.policy_id}")
    print(f"Reason: {e.message}")
    print(f"Matched rules: {e.matched_rules}")
except BudgetExceededError as e:
    print(f"Budget exceeded: ${e.current_spend:.2f} / ${e.budget_limit:.2f}")
except ConnectionError as e:
    print(f"Cannot reach Fulcrum server: {e}")
    # Handle based on failure mode
except TimeoutError:
    print("Request timed out")
```

## Integration Examples

### LangChain Integration

```python
from langchain.agents import AgentExecutor
from fulcrum import FulcrumClient

client = FulcrumClient.from_env()

def governed_agent_run(agent: AgentExecutor, query: str):
    with client.envelope(workflow_id="langchain-agent") as env:
        # Check if query is allowed
        if not env.guard("process_query", input_text=query):
            return {"error": "Query blocked by policy"}

        # Run agent with governance wrapper
        for step in agent.iter(query):
            if "tool" in step:
                tool_name = step["tool"]
                tool_input = step["tool_input"]

                # Check tool usage
                if not env.guard(tool_name, input_text=str(tool_input)):
                    env.log("tool_blocked", {"tool": tool_name})
                    continue

                env.log("tool_executed", {"tool": tool_name})

        return agent.invoke(query)
```

### LlamaIndex Integration

```python
from llama_index import VectorStoreIndex
from fulcrum import FulcrumClient

client = FulcrumClient.from_env()

def governed_query(index: VectorStoreIndex, query: str):
    with client.envelope(workflow_id="llamaindex-rag") as env:
        # Pre-query governance check
        if not env.guard("query", input_text=query):
            raise ValueError("Query not permitted")

        # Execute query
        response = index.as_query_engine().query(query)

        # Log for audit
        env.log("query_completed", {
            "query": query,
            "response_length": len(str(response)),
        })

        return response
```

### OpenAI Function Calling

```python
import openai
from fulcrum import FulcrumClient

client = FulcrumClient.from_env()

def governed_function_call(messages, tools):
    with client.envelope(workflow_id="openai-functions") as env:
        response = openai.chat.completions.create(
            model="gpt-4",
            messages=messages,
            tools=tools,
        )

        # Check function calls before execution
        for tool_call in response.choices[0].message.tool_calls or []:
            func_name = tool_call.function.name
            func_args = tool_call.function.arguments

            if not env.guard(func_name, input_text=func_args):
                env.log("function_blocked", {"function": func_name})
                continue

            # Execute approved function
            result = execute_function(func_name, func_args)
            env.log("function_executed", {
                "function": func_name,
                "success": True,
            })

        return response
```

## Cost Tracking

```python
with client.envelope(workflow_id="my-agent") as env:
    # ... agent execution ...

    # Get current cost
    cost = env.get_cost()
    print(f"Total cost: ${cost.total_usd:.4f}")
    print(f"Input tokens: {cost.input_tokens}")
    print(f"Output tokens: {cost.output_tokens}")
    print(f"LLM calls: {cost.llm_calls}")
```

## Async Support

```python
import asyncio
from fulcrum import AsyncFulcrumClient

async def main():
    client = AsyncFulcrumClient(
        host="localhost:50051",
        api_key="your-api-key"
    )

    async with client.envelope(workflow_id="async-agent") as env:
        allowed = await env.guard("action", input_text="hello")
        if allowed:
            result = await do_async_work()
            await env.log("completed", {"result": result})

asyncio.run(main())
```

## REST Clients

The SDK includes REST-based clients for managing Fulcrum resources through the Dashboard API. These clients are ideal for administrative tasks, dashboards, and automation workflows.

### PolicyClient

Manage governance policies programmatically:

```python
from fulcrum.clients import PolicyClient

policy_client = PolicyClient(
    base_url="https://your-fulcrum-dashboard.com",
    api_key="your-api-key",
    timeout=5.0  # optional, default: 5.0
)

# List all policies
policies = policy_client.list()

# List with filters
active_policies = policy_client.list(status="active", policy_type="budget")

# Get a specific policy
policy = policy_client.get("policy-id")

# Create a new policy
new_policy = policy_client.create(
    name="No PII in Responses",
    policy_type="content_filter",
    rules={"pattern": "pii_detection"},
    priority=1,
    enabled=True
)

# Update a policy
updated = policy_client.update("policy-id", name="Updated Name", priority=2)

# Enable/disable policies
policy_client.enable("policy-id")
policy_client.disable("policy-id")

# Delete a policy
policy_client.delete("policy-id")
```

### ApprovalClient

Manage human-in-the-loop approval workflows:

```python
from fulcrum.clients import ApprovalClient

approval_client = ApprovalClient(
    base_url="https://your-fulcrum-dashboard.com",
    api_key="your-api-key"
)

# List all pending approvals
pending = approval_client.list_pending()

# List with filters
approvals = approval_client.list(status="pending", workflow_id="customer-support-bot")

# Get approval details
approval = approval_client.get("approval-id")

# Approve a request
approval_client.approve("approval-id", comment="Looks safe to proceed")

# Deny a request
approval_client.deny("approval-id", comment="Contains sensitive information")
```

### BudgetClient

Manage AI spending budgets and limits:

```python
from fulcrum.clients import BudgetClient, BudgetScope, BudgetPeriod, BudgetAction

budget_client = BudgetClient(
    base_url="https://your-fulcrum-dashboard.com",
    api_key="your-api-key"
)

# List all budgets with summary
response = budget_client.list()
print(f"Total budgets: {response.summary.total}")
print(f"Active: {response.summary.active}, Exceeded: {response.summary.exceeded}")

# Filter by scope or status
tenant_budgets = budget_client.list_budgets(scope=BudgetScope.TENANT)
exceeded_budgets = budget_client.get_exceeded()
warning_budgets = budget_client.get_warnings()

# Get a specific budget
budget = budget_client.get("budget-id")

# Create a new budget
new_budget = budget_client.create(
    name="Marketing Team Monthly",
    scope=BudgetScope.TENANT,
    scope_id="tenant-123",
    limit_amount=1000,  # in dollars
    period=BudgetPeriod.MONTHLY,
    action=BudgetAction.SOFT_LIMIT,
    alert_thresholds=[50, 75, 90]
)

# Update budget
budget_client.update("budget-id", limit_amount=1500, alert_thresholds=[60, 80, 95])

# Helper methods
percentage = budget_client.calculate_percentage(budget)
is_near_limit = budget_client.is_at_threshold(budget, 80)

# Delete a budget
budget_client.delete("budget-id")
```

### MetricsClient

Access platform metrics and audit logs:

```python
from fulcrum.clients import MetricsClient
from datetime import date

metrics_client = MetricsClient(
    base_url="https://your-fulcrum-dashboard.com",
    api_key="your-api-key"  # optional for public metrics
)

# Public metrics (no auth required)
metrics = metrics_client.get_public_metrics()
print(f"Policies evaluated (24h): {metrics.policies_evaluated_24h}")
print(f"Avg latency: {metrics.avg_latency_ms}ms")
print(f"Blocked requests: {metrics.blocked_requests_24h}")

# Cognitive layer metrics
cognitive = metrics_client.get_cognitive_metrics()
print(f"Semantic Judge violations: {cognitive.semantic_judge_violations}")
print(f"Oracle savings: ${cognitive.oracle_savings}")

# Audit logs (auth required)
response = metrics_client.get_audit_logs(
    resource_type="Policy",
    start_date=date(2026, 1, 1),
    end_date=date(2026, 1, 31),
    page=1,
    page_size=50
)

# Convenience methods for audit logs
policy_logs = metrics_client.get_logs_by_resource_type("Policy")
user_logs = metrics_client.get_logs_by_user("user-id")
recent_logs = metrics_client.get_logs_by_date_range(
    date(2026, 1, 1),
    date(2026, 1, 31)
)
search_results = metrics_client.search_audit_logs("budget exceeded")
```

### Error Handling for REST Clients

All REST clients use a common error class:

```python
from fulcrum.clients import PolicyClient, FulcrumClientError

try:
    policy = policy_client.get("invalid-id")
except FulcrumClientError as e:
    print(f"Error: {e.message}")
    print(f"Status: {e.status_code}")  # e.g., 404, 401, 500
```

## Documentation

Full documentation: [https://docs.fulcrum.dev](https://docs.fulcrum.dev)

## Support

- Email: support@fulcrum.dev
- Documentation: https://docs.fulcrum.dev
- Issues: Contact your account representative