Python SDK

# pip install aroha
import anthropic
from aroha.agent import serve

@serve(
    name="My Agent",
    system_prompt="You are a helpful assistant.",
    provider="anthropic",
    model="claude-sonnet-4-6",
    memory="session",     # "none" | "session" | "persistent"
    auth="none",          # "none" | "bearer" | "aroha-sig"
)
async def my_agent(message: str, session_id: str, history: list, context: dict) -> str:
    client = anthropic.Anthropic()
    resp = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system="You are a helpful assistant.",
        messages=[{"role": "user", "content": message}],
    )
    return resp.content[0].text

if __name__ == "__main__":
    my_agent.start(port=8000)          # starts FastAPI on :8000
    # Register on Hub:
    # my_agent.register("https://my-agent.fly.dev", api_key="...")

import os
from aroha.agent import serve

@serve(
    name="Private Agent",
    auth="bearer",
    bearer_tokens={os.environ["AGENT_TOKEN"]},   # set[str] of allowed tokens
    # or: verify_bearer=lambda token: check_db(token)
)
async def private_agent(message, session_id, history, context):
    return "Hello, authenticated user!"

private_agent.start(port=8000)
# Callers: Authorization: Bearer <token>

from aroha import serve

@serve(
    name="Private Agent",
    auth="aroha-sig",
    allowed_callers=["did:aroha:alice", "did:aroha:ci-pipeline"],
    # resolve_caller_key defaults to the Aroha registry.
    # Omit allowed_callers to accept any verifiable caller (attribution only).
)
async def private_agent(message, session_id, history, context):
    return "Only listed agents reach this."

private_agent.start(port=8000)

from aroha import call_agent

reply = call_agent(
    "abc123def", "What is 2+2?",
    sign_as=("did:aroha:alice", alice_private_key),
)

from aroha import register_mcp, resolve_mcp

result = register_mcp(
    name="search-mcp",
    endpoint_url="https://mcp.example.com",
    tools=["search", "fetch"],       # tool names, or full {"name": ..., "inputSchema": ...} dicts
    api_key=os.environ["AROHA_API_KEY"],
)
print(result["didHash"])

mcp = resolve_mcp(result["didHash"])

from aroha import register_skill

register_skill(
    name="my-rag-skill",
    skill_type="rag",
    api_key=os.environ["AROHA_API_KEY"],
)

# Core (includes @serve, hub client, FastAPI server)
python -m venv .venv && source .venv/bin/activate
pip install aroha

# Framework bridges
pip install "aroha[langchain]"    # LangChain / LangGraph
pip install "aroha[crewai]"       # CrewAI
pip install "aroha[google-adk]"   # Google ADK
pip install "aroha[openai]"       # OpenAI SDK / Agents SDK
pip install "aroha[autogen]"      # AutoGen

# Everything
pip install "aroha[all]"

from aroha.bridges.langchain_v1 import from_chain, from_graph
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate

llm    = ChatAnthropic(model="claude-sonnet-4-6")
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("human",  "{message}"),
])
chain = prompt | llm

agent = from_chain(chain, name="LangChain Agent", memory="session")
agent.start(port=8000)

# For LangGraph (passes history as HumanMessage/AIMessage list):
# compiled_graph = builder.compile(checkpointer=MemorySaver())
# agent = from_graph(compiled_graph, name="LangGraph Agent")

from aroha.bridges.crewai_v1 import from_crew
from crewai import Agent, Task, Crew

researcher = Agent(
    role="Researcher",
    goal="Research the given topic",
    llm="anthropic/claude-sonnet-4-6",
)

def make_crew(message: str) -> Crew:
    task = Task(description=message, agent=researcher, expected_output="Findings")
    return Crew(agents=[researcher], tasks=[task])

agent = from_crew(make_crew, name="Research Agent")
agent.start(port=8000)

from aroha.bridges.google_adk import from_runner
from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService

adk_agent = LlmAgent(
    name="assistant",
    model="gemini-2.0-flash",
    instruction="You are a helpful assistant.",
)
runner = Runner(
    agent=adk_agent,
    app_name="my-app",
    session_service=InMemorySessionService(),
)
agent = from_runner(runner, name="Gemini Agent", memory="session")
agent.start(port=8000)

from aroha.bridges.openai_v1 import from_openai_client, from_openai_agent
from openai import OpenAI

# Plain chat completions (with session history):
agent = from_openai_client(
    OpenAI(), model="gpt-4o",
    system_prompt="You are a helpful assistant.",
    name="GPT Agent", memory="session",
)
agent.start(port=8000)

# OpenAI Agents SDK:
# from agents import Agent as OAIAgent
# agent = from_openai_agent(OAIAgent(name="…", instructions="…"))

from aroha.hub import register_agent, resolve_agent, call_agent, list_agents

# Register an agent programmatically (bypasses Studio wizard)
result = register_agent(
    manifest={"name": "My Agent", "arohaProtocolVersion": "1.0"},
    endpoint_url="https://my-agent.fly.dev",
    api_key="your-org-api-key",
)
print(result["didHash"])   # use in chat URL: /chat/<didHash>

# Resolve an agent by didHash
agent = resolve_agent("abc123def")
print(agent["endpointUrl"])

# Call an agent — resolves endpoint automatically
reply = call_agent("abc123def", "What is 2+2?", session_id="s1")
print(reply["message"])    # "4"

# List all registered agents
agents = list_agents(limit=20)

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

# generate_did returns a plain dict — use identity["key"] not identity.key
identity = generate_did("my-agent", "http://localhost:3000")

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=identity["did"],
    private_key=identity["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())

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,
)

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

identity = 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=identity["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=identity["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())

from aroha.identity import generate_did, extract_public_key

# generate_did returns a plain dict — all fields JSON-serialisable (except key objects)
identity = generate_did("travel-agent", "http://localhost:3000")
identity["did"]              # "did:aroha:travel-agent"
identity["private_key"]      # Ed25519PrivateKey  (keep secret)
identity["public_key"]       # Ed25519PublicKey
identity["public_key_bytes"] # raw 32 bytes
identity["document"]         # W3C DID Document as a plain dict (ready for JSON)

# Optional namespace — scopes DID to your org
identity = generate_did("travel-agent", "http://localhost:3000", namespace="acme")
identity["did"]   # "did:aroha:acme.travel-agent"

# Extract public key with revocation check (raises ValueError if revoked)
pub_bytes = extract_public_key(identity["document"])

# pip install aroha
import asyncio, os
from aroha.routing import ModelSession, ModelBudgetMandate, TaskType, AllocatorMode
from aroha.routing.providers.anthropic import AnthropicProvider

async def main():
    mandate = ModelBudgetMandate(
        token_budget=8_000,
        cost_budget_usd=0.10,
        quality_threshold=0.85,
        task_type=TaskType.CODE,
        expected_steps=2,          # hint for multi-step budget allocation
    )

    session = ModelSession.create(
        provider=AnthropicProvider(),
        mandate=mandate,
        allocator_mode=AllocatorMode.THOMPSON,       # learned budget allocation
        learner_path="/var/cache/aroha/learner.json", # persist learner state
        registry_url="https://registry.aroha-labs.com",
        registry_api_key=os.environ["AROHA_API_KEY"],
    )

    await session.warm_from_registry()   # pull crowd-sourced quality priors

    # Single-turn
    r = await session.chat("Refactor this function to be async.")
    print(r.content, r.model_id, r.n_escalations)

    # Multi-step with per-step budget slices
    async with session.step("retrieval", importance_weight=1.0, total_weight=3.0) as m:
        r1 = await session.chat("Extract key facts.", step_mandate=m)
    async with session.step("synthesis", importance_weight=2.0, total_weight=3.0) as m:
        r2 = await session.chat("Write a 200-word summary.", step_mandate=m)

    await session.close()   # flushes feedback signals, saves learner
    print(session.tracker.summary())
    # {
    #   "total_tokens": 2841, "tokens_remaining": 5159,
    #   "total_cost_usd": 0.01234567, "cost_remaining": 0.08765433,
    #   "n_calls": 3, "models_used": ["claude-haiku-4-5", "claude-sonnet-4-6"],
    # }

asyncio.run(main())

from aroha.routing import TaskType

TaskType.REASONING       # MMLU, HellaSwag benchmarks
TaskType.CODE            # HumanEval, MBPP benchmarks
TaskType.EXTRACTION      # structured extraction from text
TaskType.CREATIVE        # open-ended generation
TaskType.CLASSIFICATION  # labelling, intent detection
TaskType.SUMMARIZATION   # MT-Bench benchmarks

from aroha.routing import AllocatorMode

AllocatorMode.PROPORTIONAL        # budget / remaining_steps  (default)
AllocatorMode.IMPORTANCE_WEIGHTED # budget × (w_i / Σw_j)
AllocatorMode.THOMPSON            # Beta bandit — learned from observed efficiency

from aroha.mandate import (
    issue_intent_mandate,
    attenuate_to_cart,
    attenuate_to_payment,
    verify_mandate,
    SpendingConstraints,
)
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

user_key = Ed25519PrivateKey.generate()

# Issue an intent mandate — user authorises agent up to $500
intent = issue_intent_mandate(
    grantor_did="did:aroha:user",
    grantee_did="did:aroha:orchestrator",
    constraints=SpendingConstraints(
        spend_limit_usd=500.0,
        allowed_merchants=["merchant-a", "merchant-b"],
    ),
    private_key=user_key,
    ttl_seconds=3600,   # 1 hour; default is 3600
)

# Attenuate to cart — limits can only narrow, never widen
cart = attenuate_to_cart(
    parent=intent,
    grantee_did="did:aroha:provider",
    narrowed=SpendingConstraints(
        spend_limit_usd=79.99,           # ≤ 500
        allowed_merchants=["merchant-a"],
    ),
    grantor_private_key=user_key,
)

# verify_mandate returns a 3-tuple (not an object)
valid, mandate, reason = verify_mandate(cart.token, user_key.public_key())
if valid:
    print(mandate.constraints.spend_limit_usd)   # 79.99
else:
    print(f"Rejected: {reason}")

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

# 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"])

# 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)

# 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")

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=identity["did"],
    to_did="did:aroha:travel-agent",
    body={"capability": "search-flights", "params": {}},
    correlation_id=new_correlation_id(),
    private_key=identity["private_key"],   # Ed25519PrivateKey or raw bytes
    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=identity["private_key"],
    verification_method_id=f'{identity["did"]}#key-1',   # 3rd positional arg
)