Python SDK

pip install kakunin

import asyncio
from kakunin import Kakunin

async def main():
    async with Kakunin(api_key="kak_live_...") as client:
        # 1. Register the agent
        agent = await client.agents.create(
            name="TradeBot v2.1",
            model="gpt-4o",
            version="v2.1.0",
            model_hash="sha256:8f3c...2a91",
            financial_scope={
                "max_single_trade_usd": 50_000,
                "permitted_instruments": ["EUR/USD", "BTC/EUR"],
                "permitted_venues": ["euronext"],
                "leverage_permitted": False,
            },
        )

        # 2. Issue an X.509 certificate — private key never leaves KMS
        cert = await client.agents.certify(agent.id)
        print(cert.serial_number)   # "c4f9-17a2-6b8e"
        print(cert.expires_at)      # 2027-05-20T...

        # 3. Ingest a behavioral event
        event = await client.events.create(
            agent_id=agent.id,
            action_type="transaction_initiated",
            details={"amount": 840, "currency": "EUR", "venue": "euronext"},
        )
        print(event.risk_score)     # 0.12
        print(event.risk_band)      # 'low'

asyncio.run(main())

from kakunin import Kakunin

client = Kakunin(
    api_key="kak_live_...",                      # or kak_test_... for sandbox
    base_url="https://api.kakunin.ai/v1",        # default
    timeout=10.0,                                # per-request timeout in seconds
)

agent = await client.agents.create(
    name="Risk Analyser",
    model="claude-sonnet-4-6",
    version="v1.0.0",
    model_hash="sha256:...",
)

cert = await client.agents.certify(agent.id)
# cert.pem            — full X.509 PEM
# cert.serial_number  — use for verification
# cert.expires_at     — 365 days (MiCA Art. 70)

agent = await client.agents.get("agt-8f3c2a91d4")
print(agent.status)   # AgentStatus.active

agents, pagination = await client.agents.list(status="active", limit=50)

receipt = await client.agents.halt(agent.id, reason="Anomalous trade pattern detected")
print(receipt.signature)

risk = await client.agents.risk(agent.id, window_days=30)
print(risk.risk_band)         # RiskBand.high
print(risk.drift.drift_trend) # "increasing"

event = await client.events.create(
    agent_id=agent.id,
    action_type="data_access",
    chain_id="chn-xyz",            # optional — group events into a decision chain
    details={"resource": "portfolio_db", "rows_read": 1200},
)

events, pagination = await client.events.list(agent_id=agent.id, band="high")

chain = await client.chains.create(name="Trade execution T-984231")

await client.events.create(agent_id=agent.id, action_type="data_access", chain_id=chain.chain_id)
await client.events.create(agent_id=agent.id, action_type="transaction_initiated", chain_id=chain.chain_id)

closed = await client.chains.close(chain.chain_id)
print(closed.chain_hash)   # HMAC over ordered events — verifiable by anyone

# Verify a certificate serial
cert_data = await client.verify.certificate("c4f9-17a2-6b8e")
print(cert_data["status"])  # "active"

# Verify a signed message
result = await client.verify.message(
    agent_id=agent.id,
    payload={"order_id": "T-984231", "amount": 840},
    signature="base64...",
    certificate_serial="c4f9-17a2-6b8e",
)
print(result.valid)         # True

from kakunin.models import StandardsFramework

report = await client.reports.create(
    agent_id=agent.id,
    window_days=30,
    standards_frameworks=[StandardsFramework.iso_27001, StandardsFramework.nist_csf],
)
# report.status == ReportStatus.generating — poll until ready

from kakunin import Kakunin
from kakunin.integrations.scope import verify_agent_scope

client = Kakunin(api_key="kak_live_...")

@verify_agent_scope(client, agent_id="agt-123", required_scopes=["trade.execute"])
async def execute_trade(order: dict) -> dict:
    # Only runs if agent is active and has "trade.execute" in metadata["scopes"]
    ...

from langchain_core.tools import tool
from kakunin import Kakunin
from kakunin.integrations.langchain import KakuninToolGuard

client = Kakunin(api_key="kak_live_...")

@tool
def trade_executor(order: str) -> str:
    """Execute a trade order."""
    return execute_trade(order)

guarded = KakuninToolGuard(
    kakunin=client,
    agent_id="agt-123",
    tool=trade_executor,
    required_scopes=["trade.execute"],
)

# Drop into any LangChain agent or LCEL chain as a normal tool
agent = create_react_agent(llm, tools=[guarded], prompt=prompt)

from kakunin.integrations.langchain import langchain_scope_callback

guard = langchain_scope_callback(client, agent_id="agt-123")
chain = my_chain.with_config(callbacks=[guard])
# Raises ScopeViolationError at chain start if agent is not active

from kakunin import Kakunin
from kakunin.integrations.llamaindex import KakuninFunctionToolGuard
from llama_index.core.agent import ReActAgent

client = Kakunin(api_key="kak_live_...")

def portfolio_lookup(account_id: str) -> dict:
    """Fetch current portfolio positions."""
    return fetch_positions(account_id)

guarded = KakuninFunctionToolGuard(
    kakunin=client,
    agent_id="agt-123",
    fn=portfolio_lookup,
    name="portfolio_lookup",
    description="Fetch current portfolio positions for an account.",
    required_scopes=["data.read"],
)

agent = ReActAgent.from_tools([guarded], llm=llm, verbose=True)

from crewai import Task, Crew
from kakunin import Kakunin
from kakunin.integrations.crewai import KakuninCrewAgent

client = Kakunin(api_key="kak_live_...")

agent = KakuninCrewAgent(
    kakunin=client,
    agent_id="agt-123",
    required_scopes=["trade.execute"],   # optional scope guard
    # Standard CrewAI kwargs:
    role="Risk Analyst",
    goal="Assess and flag high-risk trades",
    backstory="...",
    tools=[...],
)

task = Task(description="Assess trade T-984231", agent=agent)
crew = Crew(agents=[agent], tasks=[task])
crew.kickoff()

from autogen import UserProxyAgent
from kakunin import Kakunin
from kakunin.integrations.autogen import KakuninConversableAgent

client = Kakunin(api_key="kak_live_...")

risk_engine = KakuninConversableAgent(
    kakunin=client,
    agent_id="agt-456",
    required_scopes=["chat.reply"],   # optional scope guard
    # Standard AutoGen kwargs:
    name="RiskEngine",
    system_message="You are a risk analyst.",
    llm_config={"model": "gpt-4o"},
)

user_proxy = UserProxyAgent(name="User")
user_proxy.initiate_chat(risk_engine, message="Assess trade T-984231")

from kakunin.integrations.autogen import KakuninHttpxMixin, KakuninConversableAgent

class MyAgent(KakuninHttpxMixin, KakuninConversableAgent):
    pass

agent = MyAgent(
    kakunin=client,
    agent_id="agt-456",
    cert_serial="c4f9-17a2-6b8e",
    name="RiskEngine",
    ...
)

# Outbound calls via agent.http_client automatically include the cert serial
resp = await agent.http_client.get("https://internal-service/prices")

from kakunin import Kakunin
from kakunin.integrations.langgraph import kakunin_node

client = Kakunin(api_key="kak_live_...")

@kakunin_node(client, agent_id="agt-123", action_type="data_access")
async def risk_engine_node(state: dict) -> dict:
    # your node logic
    return state

# Or wrap at graph construction time
graph.add_node("risk_engine", kakunin_node(client, agent_id="agt-123")(risk_engine_node))

@verify_agent_scope(client, agent_id="agt-123", required_scopes=["data.read"])
@kakunin_node(client, agent_id="agt-123", action_type="data_access")
async def data_ingestion_node(state: dict) -> dict:
    ...

pip install kakunin[camel]

from camel.agents import ChatAgent
from camel.models import ModelFactory
from kakunin import Kakunin
from kakunin.integrations.camel import KakuninToolkit

client = Kakunin(api_key="kak_live_...")

toolkit = KakuninToolkit(kakunin=client)
model = ModelFactory.create(model_platform=..., model_type=...)

agent = ChatAgent(
    model=model,
    tools=toolkit.get_tools(),
    system_message="You are a compliance-aware trading agent.",
)

from camel.messages import BaseMessage
from kakunin.integrations.camel import KakuninCamelAgent

agent = KakuninCamelAgent(
    kakunin=client,
    agent_id="agt-123",
    required_scopes=["trade.execute"],   # optional scope guard
    model=model,
    system_message=BaseMessage.make_assistant_message(
        role_name="TradeBot",
        content="You are a compliance-aware trading agent.",
    ),
)

user_msg = BaseMessage.make_user_message(role_name="User", content="Assess risk for trade T-984")
response = agent.step(user_msg)

from google.antigravity import Agent, LocalAgentConfig
from kakunin import Kakunin
from kakunin.integrations.google_antigravity import get_kakunin_hooks

client = Kakunin(api_key="kak_live_...")

# Automatically register all event reporting and scope verification hooks
hooks = get_kakunin_hooks(
    kakunin=client,
    agent_id="agt-123",
    tool_scopes_mapping={
        "run_command": ["system.command"],
        "edit_file": ["file.write"],
    },
)

config = LocalAgentConfig(
    hooks=hooks,
)

async with Agent(config=config) as agent:
    await agent.chat("Let's query the system status")

from swarm import Agent
from kakunin import Kakunin
from kakunin.integrations.openai_swarm import KakuninSwarm

client = Kakunin(api_key="kak_live_...")
swarm = KakuninSwarm(kakunin=client, agent_id="agt-123")

def execute_trade(symbol: str, amount: int) -> str:
    return f"Bought {amount} shares of {symbol}"

agent = Agent(
    name="Swarm Trader",
    functions=[execute_trade],
)

# Runs agent and automatically checks scopes for executed functions
response = swarm.run(
    agent=agent,
    messages=[{"role": "user", "content": "Buy 10 shares of AAPL"}],
    tool_scopes_mapping={"execute_trade": ["trade.execute"]},
)

from kakunin import Kakunin
from kakunin.integrations.openai_assistants import handle_assistants_requires_action

client = Kakunin(api_key="kak_live_...")

# Inside your run loop, when run.status == "requires_action":
tool_outputs = await handle_assistants_requires_action(
    kakunin=client,
    agent_id="agt-123",
    run=run,
    tool_scopes_mapping={
        "execute_market_trade": ["trade.execute"],
    },
    tool_funcs={
        "execute_market_trade": execute_market_trade_function,
    },
)

# Submit results back to OpenAI
openai_client.beta.threads.runs.submit_tool_outputs(
    thread_id=thread_id,
    run_id=run.id,
    tool_outputs=tool_outputs,
)

from kakunin import Kakunin, KakuninError, AuthenticationError, RateLimitError
from kakunin.exceptions import ScopeViolationError

try:
    await client.agents.certify(agent_id)
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after}s")
except AuthenticationError:
    print("Invalid API key")
except KakuninError as e:
    print(f"API error: {e}")

from kakunin.exceptions import ScopeViolationError

try:
    await guarded_fn()
except ScopeViolationError as e:
    print(e.agent_id)        # "agt-123"
    print(e.agent_status)    # "suspended"
    print(e.missing_scopes)  # ["trade.execute"]

# Required
KAK_API_KEY=kak_live_...    # or kak_test_... for sandbox