fulcrum-governance 0.1.4

Python SDK for Fulcrum - Intelligent AI Governance Platform

Fulcrum Python SDK

Intelligent AI Governance for Enterprise Agents

Installation

pip install fulcrum-governance

Quick Start

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

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

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

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

API Reference

FulcrumClient

The main client for interacting with Fulcrum.

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.

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

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

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

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

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

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

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:

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:

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:

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:

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:

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

Support

• Email: [e-mail]
• Documentation: https://docs.fulcrum.dev
• Issues: Contact your account representative