keel-ai 0.1.0

Python SDK for Keel — agent-governance compliance middleware

keel (Python SDK)

Python SDK for Keel — agent-governance compliance middleware for regulated finance. Wrap your agent's tool calls so every action is evaluated against policy before it runs.

Async-first (httpx + asyncio), fully typed (Pydantic v2, mypy --strict).

Installation

pip install keel-ai          # core SDK (KeelClient + guard)
pip install keel-ai[mcp]     # adds MCP handler (create_mcp_tool_handler)

KeelClient

import asyncio
from keel import KeelClient, KeelClientOptions, Policy, EvaluationContext, Action

policy = Policy(
    id="treasury-cap",
    name="Treasury per-transaction cap",
    version=1,
    default_action="allow",
    rules=[
        {
            "id": "block-over-50k",
            "action": "block",
            "reason": "Over per-transaction limit.",
            "condition": {
                "type": "spend_limit_per_transaction",
                "currency": "USD",
                "max_amount": 50_000,
            },
        }
    ],
)

async def main() -> None:
    client = KeelClient(KeelClientOptions(base_url="https://api.keel.dev"))
    result = await client.evaluate(
        policy,
        EvaluationContext(
            agent_id="treasury-agent",
            correlation_id="corr-1",
            timestamp="2026-06-01T12:00:00Z",
            action=Action(type="transaction", amount=10_000, currency="USD"),
        ),
    )
    print(result.decision)  # "allow"

asyncio.run(main())

guard()

Wrap async tool handlers so each call is enforced against the policy. A block raises KeelGuardBlockedError (the handler never runs); an escalate raises KeelGuardEscalatedError; an allow runs the original handler.

from keel import KeelClient, KeelClientOptions, guard, GuardOptions, Action
from keel import KeelGuardBlockedError

client = KeelClient(KeelClientOptions(base_url="https://api.keel.dev"))

async def wire_transfer(args: dict) -> str:
    return f"wired {args['amount']} to {args['counterparty']}"

guarded = guard(
    GuardOptions(
        client=client,
        policy=policy,
        agent_id="treasury-agent",
        tool_to_action=lambda name, args: Action(
            type="transaction",
            amount=args.get("amount"),
            currency=args.get("currency"),
        ),
    ),
    {"wire_transfer": wire_transfer},
)

async def run() -> None:
    try:
        out = await guarded["wire_transfer"]({"amount": 60_000, "currency": "USD"})
        print(out)
    except KeelGuardBlockedError as err:
        print(f"blocked: {err.reason} (policy {err.policy_id}, rule {err.rule_id})")

Using with MCP

If you expose your tools over the Model Context Protocol, create_mcp_tool_handler wraps an MCP tools/call handler so every invocation is enforced by guard() and guard outcomes become MCP-shaped CallToolResults with isError=True (block / escalate / governance-unreachable / unknown tool) — so the model can self-correct instead of the action slipping through.

This requires the optional mcp extra:

pip install keel[mcp]

from keel import KeelClient, KeelClientOptions, Action
from keel.mcp import McpGuardOptions, McpToolHandlerOptions, create_mcp_tool_handler
from mcp.types import CallToolRequest, CallToolResult, TextContent

client = KeelClient(KeelClientOptions(base_url="https://api.keel.dev"))

async def wire_transfer(args: dict) -> CallToolResult:
    return CallToolResult(
        content=[TextContent(type="text", text=f"wired {args['amount']}")],
    )

handle = create_mcp_tool_handler(
    McpToolHandlerOptions(
        guard=McpGuardOptions(
            client=client,
            policy=policy,
            agent_id="treasury-agent",
            tool_to_action=lambda name, args: Action(
                type="transaction", amount=args.get("amount"), currency=args.get("currency")
            ),
        ),
        tools={"wire_transfer": wire_transfer},
    )
)

# Dispatch an MCP tools/call request through Keel:
result = await handle(
    CallToolRequest(
        method="tools/call",
        params={"name": "wire_transfer", "arguments": {"amount": 60_000, "currency": "USD"}},
    )
)
# result.isError is True when the policy blocked/escalated the call.

Fail-open

Set failure_mode="open" on a low-risk Policy (serialized as failureMode) so that, if the Keel API is unreachable (KeelNetworkError / KeelEvaluateTimeoutError), the guard runs the tool unguarded and logs a guard_fail_open line to stderr instead of failing closed. Default is closed.