sunbeam-cim 1.0.0

Control in Motion SDK -- runtime governance for AI agents

Control in Motion SDK

Runtime governance enforcement for AI agents. Every tool call your agent makes is checked against your governance policies and delegation contracts before it executes.

Installation

pip install sunbeam-cim

With LangChain support:

pip install sunbeam-cim[langchain]

Quick Start

from sunbeam_cim import CIMAgent

agent = CIMAgent(
    gateway_url="http://localhost:8001",
    agent_id="my-agent-001",
    tenant_id="00000000-0000-0000-0000-000000000001",
)

# Check before acting
decision = agent.check("http_request", domain="api.example.com", method="GET")

if decision.allowed:
    response = requests.get("https://api.example.com/data")
elif decision.denied:
    print(f"Blocked: {decision.explanation}")
elif decision.requires_approval:
    print(f"Waiting for approval: {decision.approval_id}")

Environment Variables

Set these instead of passing constructor args:

export CIM_GATEWAY_URL="http://localhost:8001"
export CIM_AGENT_ID="my-agent-001"
export CIM_TENANT_ID="00000000-0000-0000-0000-000000000001"
export CIM_ENVIRONMENT="prod"

Then just:

from sunbeam_cim import CIMAgent
agent = CIMAgent()  # reads from env

Action Types

HTTP Request

decision = agent.check(
    action_type="http_request",
    domain="api.stripe.com",
    method="POST",
    body={"amount": 1000},
)

Send Message

decision = agent.check(
    action_type="send_message",
    recipient="[e-mail]",
    data_class="internal",
)

Database Query

decision = agent.check(
    action_type="query_db",
    db="production",
    table="users",
)

Decision Object

decision.allowed           # bool — True if ALLOW
decision.denied            # bool — True if DENY
decision.requires_approval # bool — True if REQUIRE_APPROVAL
decision.redacted          # bool — True if REDACT
decision.explanation       # str  — human-readable reason
decision.policy_id         # str  — which policy matched
decision.approval_id       # str  — approval request ID (if pending)
decision.redacted_fields   # list — fields that were masked

# Raise exception instead of checking manually
decision.raise_if_denied()      # raises CIMError on DENY
decision.raise_if_not_allowed() # raises CIMError on anything except ALLOW

Error Handling

from sunbeam_cim import CIMAgent, CIMError

agent = CIMAgent(...)

try:
    decision = agent.check("http_request", domain="api.example.com")
    decision.raise_if_denied()
    # safe to proceed
except CIMError as e:
    print(f"Governance blocked: {e}")
    print(f"Decision was: {e.decision.decision}")

Gateway Unreachable

If the gateway is down, the SDK returns DENY by default (fail-closed):

decision = agent.check("http_request", domain="api.example.com")
# If gateway is unreachable:
# decision.denied == True
# decision.explanation == "Gateway unreachable — action blocked by default"

LangChain Integration

Wrap a tool

from langchain.tools import tool
from sunbeam_cim import CIMAgent

agent = CIMAgent(...)

@tool
def search_web(query: str) -> str:
    """Search the web for information."""
    return requests.get(f"https://api.search.com?q={query}").text

# Wrap with governance
governed_search = agent.as_langchain_tool(
    search_web,
    action_type="http_request",
    domain="api.search.com",
)

# Use in your agent as normal — governance runs automatically

Callback handler (intercepts all tools)

from langchain.agents import AgentExecutor

executor = AgentExecutor(
    agent=llm_agent,
    tools=tools,
    callbacks=[agent.as_langchain_callback()],
)
# Every tool call now goes through governance before executing

Full ActionEvent (advanced)

from sunbeam_cim import CIMAgent, ActionEvent

agent = CIMAgent(...)

event = ActionEvent(
    tenant_id   = "00000000-...",
    agent_id    = "my-agent-001",
    action_type = "http_request",
    tool_name   = "stripe_charge",
    target      = {"domain": "api.stripe.com", "method": "POST"},
    request     = {"body": {"amount": 5000, "currency": "usd"}},
    context     = {
        "data_class": "confidential",
        "tags":       ["verified", "production-ready"],
        "user_id":    "user-123",
    },
    environment = "prod",
)

decision = agent.invoke(event)

Demo (works with local dev stack)

from sunbeam_cim import CIMAgent

agent = CIMAgent(
    gateway_url = "http://localhost:8001",
    agent_id    = "demo-agent-001",
    tenant_id   = "00000000-0000-0000-0000-000000000001",
    environment = "dev",
)

# Should ALLOW (if policy permits)
d = agent.check("http_request", domain="api.example.com")
print(d)  # Decision(ALLOW: Permitted by policy)

# Should DENY (if blocklist policy active)
d = agent.check("http_request", domain="evil.com")
print(d)  # Decision(DENY: Domain blocked by policy)