Python SDK

asyncio-native Aroha SDK with FastAPI transport, Pydantic validation, and first-class bridges for LangChain, CrewAI, and AutoGen.

Installation

# Core only
python -m venv .venv && source .venv/bin/activate  # required on macOS/Linux
pip install aroha

# With LangChain bridge
pip install "aroha[langchain]"

# With CrewAI bridge
pip install "aroha[crewai]"

# With AutoGen bridge
pip install "aroha[autogen]"

# Everything
pip install "aroha[langchain,crewai,autogen]"

ArohaServer

Minimal server

import asyncio
from aroha.transport import ArohaServer
from aroha.identity import generate_did
from aroha.messages import ArohaEnvelope, NonceRegistry

# generate_did(agent_id, endpoint) → AgentKeyPair
# AgentKeyPair has: .did, .private_key, .public_key, .public_key_bytes, .document
keypair = generate_did("my-agent", "http://localhost:3000")
my_did  = keypair.did

async def handle_message(envelope: ArohaEnvelope, respond, stream):
    if envelope["type"] == "ArohaRequest":   # envelope is a plain dict
        capability = envelope["body"]["capability"]
        params     = envelope["body"]["params"]

        if capability == "greet":
            await respond({
                "type": "ArohaResponse",
                "body": {"capability": "greet", "result": {"message": f"Hello, {params['name']}!"}},
            })

async def resolve_public_key(did: str) -> bytes | None:
    # Return sender's Ed25519 public key bytes, or None if unknown
    return None

server = ArohaServer(
    agent_did=keypair.did,
    private_key=keypair.private_key,
    port=3000,
    on_message=handle_message,
    resolve_public_key=resolve_public_key,
    dev_mode=True,   # skip crypto in development
)

asyncio.run(server.start())

Production server with middleware

from aroha.transport import ArohaServer
from aroha.middleware import rbac_middleware, rate_limit_middleware

server = ArohaServer(
    agent_did=my_did,
    private_key=private_key,
    port=3000,
    on_message=handle_message,
    resolve_public_key=resolve_public_key,
    middleware=[
        rbac_middleware(required_trust_level=2),
        rate_limit_middleware(requests_per_minute=100),
    ],
    clock_tolerance_ms=5000,
    max_body_bytes=1_048_576,
)

ArohaClient

import asyncio
from aroha.client import ArohaClient
from aroha.messages import build_envelope, new_correlation_id
from aroha.identity import generate_did

keypair = generate_did("orchestrator", "http://localhost:3001")

async def main():
    client = ArohaClient()

    # Note: first param is type_ (not type) — Python builtin conflict
    envelope = build_envelope(
        type_="ArohaRequest",          # ← type_, not type
        from_did=keypair.did,
        to_did="did:aroha:travel-agent",
        body={
            "capability": "search-flights",
            "params": {"from": "JFK", "to": "LHR", "date": "2026-08-01"},
        },
        correlation_id=new_correlation_id(),
        private_key=keypair.private_key,
    )

    response = await client.send("http://travel-agent.example.com", envelope)

    if response and response["type"] == "ArohaResponse":
        print(response["body"]["result"])  # {"flights": [...]}

asyncio.run(main())

Identity & DIDs

from aroha.identity import generate_did

# generate_did(agent_id: str, endpoint: str) → AgentKeyPair
keypair = generate_did("travel-agent", "http://localhost:3000")

# AgentKeyPair fields:
keypair.did              # "did:aroha:travel-agent"
keypair.private_key      # Ed25519PrivateKey
keypair.public_key       # Ed25519PublicKey
keypair.public_key_bytes # raw bytes
keypair.document         # DIDDocument dataclass

# DIDDocument → JSON-serialisable dict
doc_dict = keypair.document.to_dict()

Spending Mandates

from aroha.mandate import (
    issue_intent_mandate,
    attenuate_mandate,
    verify_mandate,
    SpendingConstraints,
)
from datetime import datetime, timedelta

constraints = SpendingConstraints(
    spend_limit_usd=500.0,
    allowed_capabilities=["search-flights", "reserve-flight"],
    expires_at=datetime.utcnow() + timedelta(hours=1),
)

# User authorises the orchestrator
mandate = issue_intent_mandate(
    grantor_did=user_did,
    grantee_did=orchestrator_did,
    constraints=constraints,
    private_key=user_private_key,
)

# Orchestrator delegates a sub-mandate to a sub-agent (limits can only narrow)
sub_mandate = attenuate_mandate(
    parent=mandate,
    grantee_did=flight_agent_did,
    constraints=SpendingConstraints(
        spend_limit_usd=300.0,           # ≤ 500
        allowed_capabilities=["reserve-flight"],
    ),
    private_key=orchestrator_private_key,
)

# Verify on the receiving agent
is_valid, reason = await verify_mandate(sub_mandate, expected_grantee=flight_agent_did)
print(is_valid, reason)  # True, "ok"

Saga Orchestration

from aroha.orchestrator import SagaEngine, SagaStep
import asyncio

engine = SagaEngine(client=client, orchestrator_did=my_did, private_key=private_key)

result = await engine.run([
    SagaStep(
        agent_did="did:aroha:flight-agent",
        endpoint="http://flights.example.com",
        capability="reserve-flight",
        params={"from": "JFK", "to": "LHR", "date": "2026-08-01"},
        budget_usd=300,
    ),
    SagaStep(
        agent_did="did:aroha:hotel-agent",
        endpoint="http://hotels.example.com",
        capability="reserve-hotel",
        params={"city": "London", "nights": 3},
        budget_usd=200,
    ),
])

print(result.status)          # "committed" or "compensated"
for step in result.steps:
    print(step.commit_token)  # use for billing reconciliation

LangChain Bridge

# pip install "aroha[langchain]"
from aroha.bridges.langchain_bridge import (
    aroha_capability_to_langchain_tool,
    langchain_tools_to_aroha_provider,
)
from langchain_anthropic import ChatAnthropic
from langchain.agents import create_tool_calling_agent, AgentExecutor

# Wrap an Aroha capability as a LangChain tool
flight_tool = aroha_capability_to_langchain_tool(
    capability_id="search-flights",
    endpoint="http://travel-agent.example.com",
    agent_did="did:aroha:travel-agent",
    caller_did=my_did,
    caller_private_key=private_key,
    description="Search available flights between airports",
)

# Build a LangChain agent that can call Aroha agents
llm   = ChatAnthropic(model="claude-sonnet-4-6")
agent = create_tool_calling_agent(llm, [flight_tool], prompt)
executor = AgentExecutor(agent=agent, tools=[flight_tool])

result = await executor.ainvoke({"input": "Find flights from JFK to London next Tuesday"})
print(result["output"])

CrewAI Bridge

# pip install "aroha[crewai]"
from aroha.bridges.crewai_bridge import aroha_capability_to_crewai_tool
from crewai import Agent, Task, Crew

flight_tool = aroha_capability_to_crewai_tool(
    capability_id="search-flights",
    endpoint="http://travel-agent.example.com",
    agent_did="did:aroha:travel-agent",
    caller_did=my_did,
    caller_private_key=private_key,
    description="Search available flights — params: from, to, date",
)

travel_agent = Agent(
    role="Travel Coordinator",
    goal="Find the best travel options for the user",
    tools=[flight_tool],
    llm="claude-sonnet-4-6",
)

task = Task(
    description="Find flights from JFK to LHR on 2026-08-01",
    agent=travel_agent,
)

crew = Crew(agents=[travel_agent], tasks=[task])
result = crew.kickoff()
print(result)

AutoGen Bridge

# pip install "aroha[autogen]"
from aroha.bridges.autogen_bridge import aroha_capability_to_autogen_function
from autogen import AssistantAgent, UserProxyAgent

search_flights_fn = aroha_capability_to_autogen_function(
    capability_id="search-flights",
    endpoint="http://travel-agent.example.com",
    agent_did="did:aroha:travel-agent",
    caller_did=my_did,
    caller_private_key=private_key,
)

assistant = AssistantAgent(
    name="TravelAssistant",
    llm_config={
        "functions": [search_flights_fn.schema],
        "config_list": [{"model": "claude-sonnet-4-6"}],
    },
    function_map={search_flights_fn.name: search_flights_fn.execute},
)

user_proxy = UserProxyAgent(name="User", human_input_mode="NEVER")
user_proxy.initiate_chat(assistant, message="Find me flights from JFK to London on August 1st")

Envelopes & Validation

from aroha.messages import (
    build_envelope,       # build_envelope(type_, from_did, to_did, body, correlation_id, private_key)
    validate_envelope,    # returns (bool, Optional[str]) — NOT a dict
    NonceRegistry,        # required for validate_envelope — tracks seen nonces
    new_correlation_id,
    ArohaErrorCode,
)
from aroha.crypto import sign_message   # sign_message(message, private_key, verification_method_id)

# NonceRegistry prevents replay attacks — create one per server instance
registry = NonceRegistry()

# Validate inbound envelope
valid, reason = validate_envelope(
    envelope=incoming,
    sender_public_key=sender_pub_key,   # Ed25519PublicKey object
    my_did="did:aroha:my-agent",
    nonce_registry=registry,            # required — use skip_signature=True only on trusted mesh
)
if not valid:
    raise ValueError(f"Invalid envelope: {reason}")

# Build outbound envelope (type_ with underscore — 'type' is a Python builtin)
env = build_envelope(
    type_="ArohaRequest",
    from_did=keypair.did,
    to_did="did:aroha:travel-agent",
    body={"capability": "search-flights", "params": {}},
    correlation_id=new_correlation_id(),
    private_key=keypair.private_key,
    ttl_seconds=300,                    # optional, default 300
)

# sign_message is called automatically by build_envelope.
# Only call directly if you need a standalone proof:
proof = sign_message(
    message=env,
    private_key=keypair.private_key,
    verification_method_id=f"{keypair.did}#key-1",   # 3rd positional arg, not keyword 'key_id'
)

All Modules