Metadata-Version: 2.4
Name: sunbeam-cim
Version: 1.0.0
Summary: Control in Motion SDK -- runtime governance for AI agents
Author-email: Sunbeam Control <piyush.shandilya@sunbeamcontrol.com>
License-Expression: MIT
Project-URL: Homepage, https://sunbeamcontrol.com
Project-URL: Documentation, https://sunbeamcontrol.com/cim
Requires-Python: >=3.9
Description-Content-Type: text/markdown
Requires-Dist: requests>=2.28.0
Provides-Extra: langchain
Requires-Dist: langchain>=0.1.0; extra == "langchain"
Provides-Extra: async
Requires-Dist: httpx>=0.24.0; extra == "async"
Provides-Extra: dev
Requires-Dist: pytest; extra == "dev"
Requires-Dist: pytest-mock; extra == "dev"
Requires-Dist: responses; extra == "dev"

# 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

```bash
pip install sunbeam-cim
```

With LangChain support:
```bash
pip install sunbeam-cim[langchain]
```

## Quick Start

```python
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:

```bash
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:
```python
from sunbeam_cim import CIMAgent
agent = CIMAgent()  # reads from env
```

## Action Types

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

### Send Message
```python
decision = agent.check(
    action_type="send_message",
    recipient="customer@example.com",
    data_class="internal",
)
```

### Database Query
```python
decision = agent.check(
    action_type="query_db",
    db="production",
    table="users",
)
```

## Decision Object

```python
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

```python
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):

```python
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

```python
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)

```python
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)

```python
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)

```python
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)
```
