fulcrum-governance 0.1.5

Python SDK for Fulcrum - Intelligent AI Governance Platform

Fulcrum Python SDK

Intelligent AI Governance for Enterprise Agents

Installation

pip install fulcrum-governance

Quick Start

Hosted Fulcrum uses the REST API and X-API-Key authentication:

import os
from fulcrum import (
    PolicyClient,
    EnvelopeClient,
    CreateEnvelopeRequest,
    UpdateEnvelopeStatusRequest,
)

api_base = os.getenv("FULCRUM_API_BASE", "https://api.fulcrumlayer.io")
api_key = os.environ["FULCRUM_API_KEY"]

policies = PolicyClient(base_url=api_base, api_key=api_key).list()
print(f"Visible policies: {len(policies)}")

envelopes = EnvelopeClient(base_url=api_base, api_key=api_key)
envelope = envelopes.create(CreateEnvelopeRequest(
    workflow_id="customer-support-bot",
    metadata={"example": "python-sdk"},
))
envelopes.update_status(UpdateEnvelopeStatusRequest(
    envelope_id=envelope.envelope_id,
    status="RUNNING",
))

# Run your agent work here, then close the envelope.
envelopes.update_status(UpdateEnvelopeStatusRequest(
    envelope_id=envelope.envelope_id,
    status="COMPLETED",
))

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_API_KEY="your-api-key"
export FULCRUM_API_BASE="https://api.fulcrumlayer.io"
export FULCRUM_TIMEOUT_MS="500"

The lower-level FulcrumClient gRPC API is still available for compatible self-hosted or direct gRPC deployments. The hosted first-run path uses the REST clients above.

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

List governance policies visible to your API key. Create and edit hosted policies in the Fulcrum dashboard before using the API-key gateway.

from fulcrum.clients import PolicyClient

policy_client = PolicyClient(
    base_url="https://api.fulcrumlayer.io",
    api_key="your-api-key",
    timeout_ms=5000,
)

# List all policies
policies = policy_client.list()

# List with filters
active_policies = policy_client.list()

# Get a specific policy
policy = policy_client.get_policy("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

Release & Publishing

Release agents: before publishing a new version of fulcrum-governance to PyPI, run through the authoritative pre-publish checklist:

• PYPI_PUBLICATION_CHECKLIST.md — package structure, metadata, build/upload steps, and post-publish smoke tests.

The checklist is the canonical pre-publish gate for this SDK; do not skip it or duplicate ad-hoc release steps elsewhere.

Support

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