Fulcrum Python SDK

Intelligent AI Governance for Enterprise Agents

Installation

pip install fulcrum-governance

Testing

Run the SDK unit and functional suite from the repository root:

uv run --isolated --directory sdk/python --extra dev pytest -q

The two live gRPC integration tests are skipped unless a Fulcrum server is explicitly provisioned. To run them, set both the opt-in and the server address:

FULCRUM_SDK_RUN_INTEGRATION=1 \
FULCRUM_SDK_INTEGRATION_ADDR=127.0.0.1:50051 \
uv run --isolated --directory sdk/python --extra dev pytest -q tests/test_integration.py

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
• Explicit Enforcement Modes: enforcement blocks indeterminate checks by default; audit-only is opt-in

Configuration

Client Options

from fulcrum import FulcrumClient

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="FAIL_CLOSED",          # default: blocks CHECK_INDETERMINATE
    timeout_ms=500,                  # Request timeout in milliseconds
    enable_tls=True,                 # Enable TLS encryption
    ca_cert_path="/path/to/ca.crt",  # Custom CA certificate (optional)
)

Use on_failure="FAIL_OPEN" only for legacy, explicit audit-only observation. It reports AUDIT_ONLY CHECK_INDETERMINATE and cannot make an enforcing decision. New instrumentation code should prefer the named mode below.

Instrumentation modes

from fulcrum.instrumentation import auto_govern

# Default enforcement: a missing, expired, malformed, timed-out, or cancelled
# required check blocks the pending action as CHECK_INDETERMINATE.
auto_govern.activate(api_key=api_key, tenant_id="tenant-123")

# Explicit observation-only mode: it records would-have-blocked results but
# does not deny or forward calls itself.
auto_govern.activate(
    api_key=api_key,
    tenant_id="tenant-123",
    enforcement_mode="audit_only",
)

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

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

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: support@fulcrum.dev
• Documentation: https://docs.fulcrum.dev
• Issues: Contact your account representative