Skip to content

API Reference

Auto-generated from the source via mkdocstrings. The top-level package re-exports the most-used names — from claudeway import Swarm, ConsensusReceipt, Ed25519Backend is all you need for most apps.

For examples, see examples/.


claudeway — top-level package

claudeway

Claudeway Core - Multi-agent coordination that actually agrees.

A clean, efficient implementation of multi-agent Claude orchestration. Real consensus, real decomposition, real concurrency.

Agent

Agent(config: AgentConfig, api_key: str | None = None)

A Claude agent - can reason, use tools, and collaborate.

Minimal implementation. No unnecessary abstractions.

think async

think(input_message: str) -> str

Process an input and generate a response.

Core thinking loop - uses Claude to reason about the input. Handles tool use automatically if Claude requests it.

reset

reset() -> None

Clear conversation history.

get_state

get_state() -> dict[str, Any]

Get agent state for inspection.

AgentConfig dataclass

AgentConfig(name: str, role: str, instructions: str, model: str = 'claude-3-5-sonnet-20241022', temperature: float = 0.7, max_tokens: int = 4096, tools: list[Any] = list())

Configuration for an agent.

Message dataclass

Message(role: str, content: str, timestamp: datetime = datetime.utcnow(), tool_calls: list[dict] = list(), tool_results: list[dict] = list())

A message in an agent conversation.

to_api_format

to_api_format() -> dict[str, Any]

Convert to Anthropic API format.

ConsensusResult dataclass

ConsensusResult(final_answer: str, method: str, agent_count: int, responses: list[AgentResponse], agreement: float = 0.0, rounds: int = 1, disagreed: bool = False)

The outcome of a consensus round.

ConsensusStrategy

Bases: ABC

How a swarm turns many agent responses into one answer.

Implementations receive the responses collected for the current round and return a ConsensusResult. They may request additional rounds (Debate) by calling back into the swarm for revised responses.

resolve abstractmethod async

resolve(responses: list[AgentResponse], swarm: Swarm) -> ConsensusResult

Turn responses into a ConsensusResult.

Debate

Debate(agreement_threshold: float = 0.8)

Bases: ConsensusStrategy

One revision round: agents see peers' answers and revise.

2N calls total. Use for genuinely hard questions where WeightedVote disagrees. The final answer is the highest-confidence revised response.

agreement_threshold: if the first round's agreement score is at or above this, the debate is considered settled and the costly revision round is skipped.

WeightedVote

Bases: ConsensusStrategy

Pick the highest-confidence answer, surfacing disagreement.

One round, N calls. Default for cost reasons.

Coordinator

Coordinator(config: CoordinatorConfig, api_key: str | None = None)

Hierarchical coordinator - manages specialist agents.

Unlike the swarm (peer-to-peer consensus), this is hierarchical: one coordinator plans and delegates, specialists execute parts.

add_sub_agent

add_sub_agent(agent_id: str, agent: Agent) -> None

Add a specialist agent that this coordinator can delegate to.

add_sub_swarm

add_sub_swarm(swarm_id: str, swarm: Swarm) -> None

Add a sub-swarm that this coordinator can delegate to.

coordinate async

coordinate(task: Task) -> Task

Coordinate a complex task by breaking it into sub-tasks.

  1. Decompose the task (real JSON parse).
  2. Assign sub-tasks to specialists.
  3. Execute respecting dependencies (parallel where independent).
  4. Synthesize a final answer.

CoordinatorConfig

CoordinatorConfig(name: str = 'Coordinator', role: str = 'Task Coordinator and Manager', instructions: str = 'You are a Task Coordinator responsible for:\n\n1. Breaking down complex tasks into manageable sub-tasks\n2. Identifying which specialist agents should handle each sub-task\n3. Managing dependencies between sub-tasks\n4. Synthesizing sub-task results into a final answer\n\nWhen given a task, analyze it and create a structured plan with sub-tasks.\nBe specific about what each sub-task requires and who should handle it.', model: str = 'claude-3-5-sonnet-20241022', temperature: float = 0.7, max_tokens: int = 4096)

Bases: AgentConfig

Configuration for a coordinator agent.

SubTask dataclass

SubTask(id: str, parent_task: str, description: str, assigned_to: str | None = None, specialist_role: str | None = None, dependencies: list[str] = list(), status: str = 'pending', result: Any = None)

A sub-task created by decomposition.

Runtime

Runtime()

Claudeway Runtime - manages agent processes.

Responsibilities: - Spawn and manage agent processes - Route tasks to appropriate agents - Monitor health and restart failed agents

start async

start() -> None

Start the runtime.

stop async

stop() -> None

Stop the runtime and all agents.

spawn_agent

spawn_agent(config: AgentConfig, agent_id: str | None = None) -> str

Spawn a new agent process.

Returns the agent ID.

create_swarm

create_swarm(config: SwarmConfig, swarm_id: str | None = None) -> str

Create a new swarm from configuration.

Returns the swarm ID.

submit_task async

submit_task(swarm_id: str, task: Task) -> Task

Submit a task to a swarm for processing.

Returns the completed task with results.

get_status

get_status() -> dict[str, Any]

Get runtime status.

ConsensusReceipt dataclass

ConsensusReceipt(payload: dict[str, Any], algorithm: str = '', public_key: str = '', signature: str = '', payload_hash: str = '', signed_at: str = '', metadata: dict[str, Any] = dict())

A signed (or signable) consensus attestation.

payload: the canonical consensus facts (answer, agents, agreement...). algorithm: which SignatureBackend produced the signature. public_key: hex-encoded verification key of the signer. signature: hex-encoded signature over sha256(canonical_json(payload)). payload_hash: hex sha256 of the canonical payload, for quick comparison. signed_at: UTC timestamp of signing. metadata: free-form, transport-specific hints (not signed).

from_result classmethod

from_result(result: ConsensusResult, swarm_name: str = '', task_id: str = '') -> ConsensusReceipt

Build an unsigned receipt from a ConsensusResult.

canonical_payload

canonical_payload() -> str

Return the exact string that signatures are computed over.

Ed25519Backend

Ed25519Backend()

Bases: SignatureBackend

Default signature backend. Ed25519 over the cryptography lib (already a dep).

No new native dependencies, works on Windows/Linux/Mac, fast, compact (64-byte signatures). Not Nostr-native — use NostrBackend for Buzz interop.

SignatureBackend

Bases: ABC

Swappable crypto for signing receipts.

Implementations: Ed25519Backend (default), and (opt-in) a Nostr/secp256k1 backend. The interface is intentionally tiny so a future post-quantum backend (ML-DSA / SLH-DSA) drops in without changes elsewhere.

generate_keypair abstractmethod

generate_keypair() -> tuple[str, str]

Return (private_key_hex, public_key_hex).

sign abstractmethod

sign(message: bytes, private_key_hex: str) -> str

Sign message bytes; return hex signature.

verify abstractmethod

verify(message: bytes, signature_hex: str, public_key_hex: str) -> bool

Return True iff signature is valid for message under public_key.

sign_receipt

sign_receipt(receipt: ConsensusReceipt, private_key_hex: str) -> ConsensusReceipt

Sign a receipt in place and return it.

verify_receipt

verify_receipt(receipt: ConsensusReceipt) -> bool

Verify a signed receipt. Unsigned or tampered receipts return False.

MLDSABackend

MLDSABackend()

Bases: SignatureBackend

Post-quantum signature backend. ML-DSA-65 (FIPS 204) over pure Python.

Same interface as Ed25519Backend: generate_keypair() returns (private_key_hex, public_key_hex), sign_receipt()/verify_receipt() are inherited from SignatureBackend and work unchanged. Receipts signed via this backend self-tag algorithm="mldsa65".

Signatures are ~3.3KB (vs 64 bytes for Ed25519) — the cost of post-quantum security today. Worth it for long-lived attestations (M&A diligence records, audit logs) that must stay verifiable across the transition to cryptographically relevant quantum hardware.

AgentResponse dataclass

AgentResponse(agent_name: str, answer: str = '', content: str = '', confidence: float = 0.5, reasoning: str = '', round: int = 1, timestamp: datetime = datetime.utcnow())

A response from an agent in the swarm.

Swarm

Swarm(config: SwarmConfig, api_key: str | None = None, consensus: ConsensusStrategy | None = None, on_event: OnEvent | None = None)

A swarm of Claude agents working together.

Coordinates: - Concurrent response collection (asyncio.gather) - Pluggable consensus (see core.consensus) - Multi-round debate when agents disagree

process async

process(task: Task) -> Task

Process a task through the swarm.

  1. Collect responses from all agents concurrently.
  2. Run consensus to determine the final answer.

get_state

get_state() -> dict[str, Any]

Get swarm state for inspection.

SwarmConfig dataclass

SwarmConfig(name: str, description: str, agents: list[AgentConfig] = list(), topology: str = 'hierarchical_mesh', consensus_method: str = 'weighted_vote', max_task_tokens: int = 100000)

Configuration for a swarm of agents.

Task dataclass

Task(id: str, description: str, input_data: dict[str, Any], status: str = 'pending', result: Any = None, created_at: datetime = datetime.utcnow())

A task for the swarm to work on.

claudeway.swarm — concurrent agent collection

The Swarm runs N Claude agents concurrently (asyncio.gather, not serial) and aggregates their answers through a pluggable ConsensusStrategy — see claudeway.consensus below. A swarm isn't "N agents that all answer and we pick the first" — it's N agents whose answers are aggregated with disagreement surfaced explicitly.

swarm

Claudeway Swarm - Multi-agent coordination with real consensus.

This is the differentiator. A swarm isn't "N agents that all answer and we pick the first" — it's N agents whose answers are aggregated through a pluggable ConsensusStrategy (WeightedVote by default, Debate for hard cases), with disagreement surfaced explicitly rather than papered over.

Agents run concurrently: a 3-agent swarm issues all 3 Claude calls in parallel via asyncio.gather, not serially.

SwarmConfig dataclass

SwarmConfig(name: str, description: str, agents: list[AgentConfig] = list(), topology: str = 'hierarchical_mesh', consensus_method: str = 'weighted_vote', max_task_tokens: int = 100000)

Configuration for a swarm of agents.

Task dataclass

Task(id: str, description: str, input_data: dict[str, Any], status: str = 'pending', result: Any = None, created_at: datetime = datetime.utcnow())

A task for the swarm to work on.

AgentResponse dataclass

AgentResponse(agent_name: str, answer: str = '', content: str = '', confidence: float = 0.5, reasoning: str = '', round: int = 1, timestamp: datetime = datetime.utcnow())

A response from an agent in the swarm.

Swarm

Swarm(config: SwarmConfig, api_key: str | None = None, consensus: ConsensusStrategy | None = None, on_event: OnEvent | None = None)

A swarm of Claude agents working together.

Coordinates: - Concurrent response collection (asyncio.gather) - Pluggable consensus (see core.consensus) - Multi-round debate when agents disagree

process async

process(task: Task) -> Task

Process a task through the swarm.

  1. Collect responses from all agents concurrently.
  2. Run consensus to determine the final answer.

get_state

get_state() -> dict[str, Any]

Get swarm state for inspection.

claudeway.consensus — pluggable agreement

Two strategies ship:

  • WeightedVote (default): aggregate per-agent responses weighted by reported confidence. Cheap, one round.
  • Debate: agents see peers' answers and revise once. More calls, higher-quality agreement on hard questions. Early-exits when agents already agree (cost-guarded).

Subclass ConsensusStrategy to add your own (tournament, human-in-loop, etc.).

consensus

Claudeway Consensus - How multiple agents reach agreement.

The core differentiator. Frameworks like LangGraph make you wire coordination by hand; CrewAI's coordination is shallow. Here, consensus is a first-class, pluggable primitive: give the swarm agents and a task, get back an answer the agents actually agreed on — with disagreement surfaced, not hidden.

Two strategies ship
  • WeightedVote (default): aggregate per-agent responses weighted by the confidence each agent reports. Cheap, one round.
  • Debate: agents see each other's answers and revise once. More calls, higher-quality agreement on hard questions.

Implement ConsensusStrategy to add your own (e.g. tournament, human-in-loop).

ParsedResponse dataclass

ParsedResponse(answer: str, confidence: float = 0.5, reasoning: str = '')

An agent response split into the answer and the structured meta.

ConsensusResult dataclass

ConsensusResult(final_answer: str, method: str, agent_count: int, responses: list[AgentResponse], agreement: float = 0.0, rounds: int = 1, disagreed: bool = False)

The outcome of a consensus round.

ConsensusStrategy

Bases: ABC

How a swarm turns many agent responses into one answer.

Implementations receive the responses collected for the current round and return a ConsensusResult. They may request additional rounds (Debate) by calling back into the swarm for revised responses.

resolve abstractmethod async

resolve(responses: list[AgentResponse], swarm: Swarm) -> ConsensusResult

Turn responses into a ConsensusResult.

WeightedVote

Bases: ConsensusStrategy

Pick the highest-confidence answer, surfacing disagreement.

One round, N calls. Default for cost reasons.

Debate

Debate(agreement_threshold: float = 0.8)

Bases: ConsensusStrategy

One revision round: agents see peers' answers and revise.

2N calls total. Use for genuinely hard questions where WeightedVote disagrees. The final answer is the highest-confidence revised response.

agreement_threshold: if the first round's agreement score is at or above this, the debate is considered settled and the costly revision round is skipped.

parse_structured_output

parse_structured_output(raw: str) -> ParsedResponse

Parse the structured tail of an agent response.

Agents are prompted to emit

...the substantive answer... 0.0-1.0 why

Tolerant of three failure modes
  • No tags at all: whole raw text becomes the answer.
  • Open-but-not-closed (small models truncating at max_tokens): everything after the opening tag is the answer.
  • Missing confidence/reasoning: defaults kick in (0.5 / "").

claudeway.coordinator — hierarchical decomposition

The coordinator takes a high-level task, asks a planner agent to decompose it into a JSON plan with dependencies, then runs specialists in dependency-respecting parallel order and synthesizes the result. This is the non-consensus execution path: useful when you have specialists that don't need to agree, just contribute their slice.

coordinator

Claudeway Coordinator - Hierarchical task decomposition.

A "manager" agent breaks a complex task into sub-tasks, assigns each to the right specialist, and executes them in dependency order — running independent sub-tasks in parallel. Then it synthesizes a final answer from the parts.

This is real decomposition: the coordinator's JSON plan is parsed and honored, not replaced with 3 hardcoded subtasks.

SubTask dataclass

SubTask(id: str, parent_task: str, description: str, assigned_to: str | None = None, specialist_role: str | None = None, dependencies: list[str] = list(), status: str = 'pending', result: Any = None)

A sub-task created by decomposition.

CoordinatorConfig

CoordinatorConfig(name: str = 'Coordinator', role: str = 'Task Coordinator and Manager', instructions: str = 'You are a Task Coordinator responsible for:\n\n1. Breaking down complex tasks into manageable sub-tasks\n2. Identifying which specialist agents should handle each sub-task\n3. Managing dependencies between sub-tasks\n4. Synthesizing sub-task results into a final answer\n\nWhen given a task, analyze it and create a structured plan with sub-tasks.\nBe specific about what each sub-task requires and who should handle it.', model: str = 'claude-3-5-sonnet-20241022', temperature: float = 0.7, max_tokens: int = 4096)

Bases: AgentConfig

Configuration for a coordinator agent.

Coordinator

Coordinator(config: CoordinatorConfig, api_key: str | None = None)

Hierarchical coordinator - manages specialist agents.

Unlike the swarm (peer-to-peer consensus), this is hierarchical: one coordinator plans and delegates, specialists execute parts.

add_sub_agent

add_sub_agent(agent_id: str, agent: Agent) -> None

Add a specialist agent that this coordinator can delegate to.

add_sub_swarm

add_sub_swarm(swarm_id: str, swarm: Swarm) -> None

Add a sub-swarm that this coordinator can delegate to.

coordinate async

coordinate(task: Task) -> Task

Coordinate a complex task by breaking it into sub-tasks.

  1. Decompose the task (real JSON parse).
  2. Assign sub-tasks to specialists.
  3. Execute respecting dependencies (parallel where independent).
  4. Synthesize a final answer.

claudeway.agent — a single Claude agent

Wraps AsyncAnthropic with a tool-use loop. Usually constructed indirectly via SwarmConfig.agents, but exposed for callers who want a single agent without a swarm.

agent

Claudeway Agent - A Claude agent that can think and act.

Built on Anthropic's Agent SDK with clean abstractions.

Message dataclass

Message(role: str, content: str, timestamp: datetime = datetime.utcnow(), tool_calls: list[dict] = list(), tool_results: list[dict] = list())

A message in an agent conversation.

to_api_format

to_api_format() -> dict[str, Any]

Convert to Anthropic API format.

AgentConfig dataclass

AgentConfig(name: str, role: str, instructions: str, model: str = 'claude-3-5-sonnet-20241022', temperature: float = 0.7, max_tokens: int = 4096, tools: list[Any] = list())

Configuration for an agent.

Agent

Agent(config: AgentConfig, api_key: str | None = None)

A Claude agent - can reason, use tools, and collaborate.

Minimal implementation. No unnecessary abstractions.

think async

think(input_message: str) -> str

Process an input and generate a response.

Core thinking loop - uses Claude to reason about the input. Handles tool use automatically if Claude requests it.

reset

reset() -> None

Clear conversation history.

get_state

get_state() -> dict[str, Any]

Get agent state for inspection.

claudeway.signing — verifiable, tamper-evident receipts

This is the moat. A ConsensusResult isn't just "an answer" — it's an attestation: a canonical payload signed by a key, verifiable by anyone, with the signature decoupled from how the receipt is transported.

  • ConsensusReceipt: transport-agnostic canonical payload.
  • SignatureBackend (ABC): swappable crypto. Ed25519Backend ships by default (no new deps, works everywhere). A future ML-DSA post-quantum backend drops in without touching consensus code.
  • canonical_json: deterministic serialization. Two processes producing the same receipt MUST hash identically so signatures verify across machines and languages.

signing

Claudeway signing - verifiable, tamper-evident consensus receipts.

This is the moat. A ConsensusResult isn't just "an answer" — it's an attestation: a canonical payload signed by a key, verifiable by anyone, with the signature decoupled from how the receipt is transported.

Design (future-proof): - ConsensusReceipt: transport-agnostic canonical payload. - SignatureBackend (interface): swappable crypto. Ed25519 ships by default (no new deps, works everywhere). Nostr/secp256k1 is an opt-in extra for Buzz interop. A future ML-DSA post-quantum backend drops in without touching consensus code. - Transports: the same signed receipt can be emitted as a plain JSON receipt, a Nostr event (kind 30078, NIP-78 app data), or a W3C Verifiable Credential. Signing is independent of transport.

Why decouple: the W3C AI Agent Protocol CG is converging on VC/DID for agent attestation; Nostr is one transport; post-quantum signatures are coming. Hardwiring any one of these would force a rewrite. This layer won't.

ConsensusReceipt dataclass

ConsensusReceipt(payload: dict[str, Any], algorithm: str = '', public_key: str = '', signature: str = '', payload_hash: str = '', signed_at: str = '', metadata: dict[str, Any] = dict())

A signed (or signable) consensus attestation.

payload: the canonical consensus facts (answer, agents, agreement...). algorithm: which SignatureBackend produced the signature. public_key: hex-encoded verification key of the signer. signature: hex-encoded signature over sha256(canonical_json(payload)). payload_hash: hex sha256 of the canonical payload, for quick comparison. signed_at: UTC timestamp of signing. metadata: free-form, transport-specific hints (not signed).

from_result classmethod

from_result(result: ConsensusResult, swarm_name: str = '', task_id: str = '') -> ConsensusReceipt

Build an unsigned receipt from a ConsensusResult.

canonical_payload

canonical_payload() -> str

Return the exact string that signatures are computed over.

SignatureBackend

Bases: ABC

Swappable crypto for signing receipts.

Implementations: Ed25519Backend (default), and (opt-in) a Nostr/secp256k1 backend. The interface is intentionally tiny so a future post-quantum backend (ML-DSA / SLH-DSA) drops in without changes elsewhere.

generate_keypair abstractmethod

generate_keypair() -> tuple[str, str]

Return (private_key_hex, public_key_hex).

sign abstractmethod

sign(message: bytes, private_key_hex: str) -> str

Sign message bytes; return hex signature.

verify abstractmethod

verify(message: bytes, signature_hex: str, public_key_hex: str) -> bool

Return True iff signature is valid for message under public_key.

sign_receipt

sign_receipt(receipt: ConsensusReceipt, private_key_hex: str) -> ConsensusReceipt

Sign a receipt in place and return it.

verify_receipt

verify_receipt(receipt: ConsensusReceipt) -> bool

Verify a signed receipt. Unsigned or tampered receipts return False.

Ed25519Backend

Ed25519Backend()

Bases: SignatureBackend

Default signature backend. Ed25519 over the cryptography lib (already a dep).

No new native dependencies, works on Windows/Linux/Mac, fast, compact (64-byte signatures). Not Nostr-native — use NostrBackend for Buzz interop.

canonical_json

canonical_json(data: Any) -> str

Deterministic JSON for signing.

Sorted keys, no extra whitespace, no non-ASCII escapes, fixed datetime rendering. Two processes producing the same receipt MUST hash identically so signatures verify across machines and languages.

default_backend

default_backend() -> SignatureBackend

The backend used when callers don't specify one.

claudeway.transports — JSON, W3C VC, Nostr NIP-78

See Transports for the design rationale and when to pick each.

transports

Claudeway transports - emit a signed ConsensusReceipt over a wire format.

Signing (core.signing) is independent of transport. Here we render the same signed receipt into formats downstream systems consume:

  • JSONReceipt: plain JSON, the simplest verifiable artifact.
  • NostrEvent (NIP-01/NIP-78): a kind-30078 addressable event carrying the receipt, signed per the Nostr spec (Schnorr over secp256k1). Drops into a Buzz room or any Nostr relay.
  • VerifiableCredential: W3C VC v2.0-compatible envelope (the direction the W3C AI Agent Protocol CG is standardizing on).

Nostr requires secp256k1+Schnorr (BIP-340), which needs a native lib. It's imported lazily so the core stays dependency-free; install claudeway[nostr] (pynostr / coincurve) to enable NostrEvent.

NostrEvent dataclass

NostrEvent(id: str, pubkey: str, created_at: int, kind: int, tags: list[list[str]], content: str, sig: str)

A minimal NIP-01 event (enough to publish to a relay).

to_json_receipt

to_json_receipt(receipt: ConsensusReceipt) -> dict[str, Any]

The plain, self-describing JSON form of a receipt.

to_verifiable_credential

to_verifiable_credential(receipt: ConsensusReceipt, issuer_did: str = '') -> dict[str, Any]

Wrap a receipt as a W3C VC v2.0-compatible VerifiableCredential.

The receipt's signature becomes the proof. issuer_did is optional; when present it's recorded as the issuer (a DID per the W3C AI Agent Protocol direction). This is an envelope — full VC issuer/holder/verifier semantics are out of scope here, but the shape is standards-aligned.

to_nostr_event

to_nostr_event(receipt: ConsensusReceipt, private_key_hex: str, created_at: int | None = None, d_tag: str = 'claudeway-consensus') -> NostrEvent

Render a signed receipt as a Nostr kind-30078 event.

The receipt itself is already signed by its own backend (Ed25519 by default); the Nostr event signature is an additional transport-level Schnorr signature that relays require. Requires a Nostr (secp256k1) private key — distinct from the receipt's signing key by design, so a relay's key can rotate without invalidating receipts.

Lazily imports a BIP-340-capable library. Install claudeway[nostr].

to_log_anchor_event

to_log_anchor_event(anchor: Any, private_key_hex: str, log_name: str = '', created_at: int | None = None) -> NostrEvent

Render a transparency-log anchor as a Nostr kind-30078 event.

The anchor commits to (tree_size, root) at a moment in time. Publishing it as a Nostr event makes the log's history globally visible: anyone watching the relay sees every root Claudeway has published, and a missing or replaced anchor is detectable (day-2 consistency proofs make this rigorous; day 1 relies on Nostr's append-only-on-relay semantics + the operator's reputation).

log_name is the logical log this anchor belongs to (e.g. "claudeway-canonical"). It goes in the event content, not the d-tag — the d-tag scopes by event type, the log_name scopes within that.

Reuses the existing _nostr_pubkey / _nostr_sign helpers — no new crypto. The anchor's Nostr signature is a transport-level Schnorr signature distinct from any Ed25519 receipt signature, by design.

claudeway.runtime — process runner

A thin async runtime helper. Most users won't touch this directly.

runtime

Claudeway Runtime - Manages long-running agent processes.

Spawns, monitors, and restarts agents as needed. Clean process supervision without over-engineering.

AgentStatus

Bases: Enum

Status of an agent process.

AgentProcess dataclass

AgentProcess(id: str, agent: Agent, status: AgentStatus = AgentStatus.STARTING, created_at: datetime = datetime.utcnow(), last_activity: datetime = datetime.utcnow(), task_queue: Queue[Task] = asyncio.Queue(), process_task: Task | None = None)

A running agent process.

start async

start() -> None

Start the agent process loop.

stop async

stop() -> None

Stop the agent process.

enqueue_task

enqueue_task(task: Task) -> None

Add a task to this agent's queue.

Runtime

Runtime()

Claudeway Runtime - manages agent processes.

Responsibilities: - Spawn and manage agent processes - Route tasks to appropriate agents - Monitor health and restart failed agents

start async

start() -> None

Start the runtime.

stop async

stop() -> None

Stop the runtime and all agents.

spawn_agent

spawn_agent(config: AgentConfig, agent_id: str | None = None) -> str

Spawn a new agent process.

Returns the agent ID.

create_swarm

create_swarm(config: SwarmConfig, swarm_id: str | None = None) -> str

Create a new swarm from configuration.

Returns the swarm ID.

submit_task async

submit_task(swarm_id: str, task: Task) -> Task

Submit a task to a swarm for processing.

Returns the completed task with results.

get_status

get_status() -> dict[str, Any]

Get runtime status.

claudeway.server — MCP server

Exposes reach_consensus and verify_consensus as MCP tools, so any MCP-capable agent (Claude Code, Cursor, Goose, Buzz rooms) can call consensus without learning the SDK.

Install with pip install claudeway[mcp], then:

claudeway-mcp            # stdio — for Claude Code, Cursor
claudeway-mcp --http     # HTTP/SSE — for remote agents

server

Claudeway MCP server — expose consensus as a tool to any MCP client.

This is the distribution play. Run this server and any MCP-capable agent (Claude Code, Goose, Buzz rooms, Cursor) gains a reach_consensus tool: hand it a question + N specialist perspectives, get back a signed, verifiable agreement. No framework to learn, no graphs to wire.

Run

pip install claudeway[mcp] claudeway-mcp # stdio transport (Claude Code, Cursor) claudeway-mcp --http # HTTP/SSE transport (remote agents)

The server is transport-agnostic; FastMCP picks the transport from how it's invoked. Tools are async and reuse the claudeway core directly.

main

main() -> None

Entry point for the claudeway-mcp console script.

claudeway.adapters.langgraph — LangGraph integration

See Adapters for usage and design notes.

langgraph

LangGraph adapter — drop Claudeway consensus into a StateGraph.

LangGraph is the orchestration framework that "makes you wire coordination by hand." This adapter is the seam: one node gets a signed agreement the agents actually reached — no consensus wiring on the user's side.

Two entry points, both compile-once (per LangGraph forum guidance: never compile inside a node, never return a subgraph from a node):

  • make_consensus_node(swarm) -> async node function For users who own the graph. add_node() it into their own StateGraph.

  • build_consensus_graph(swarm) -> CompiledStateGraph Prebuilt subgraph: {"question": str} -> signed agreement out. Use standalone, or add_node() the compiled graph into a parent graph.

LangGraph is imported lazily so core claudeway stays dependency-free. Install it yourself: pip install langgraph.

ConsensusState

Bases: TypedDict

LangGraph-native state shape for a consensus node.

messages accumulates (add_messages reducer) so the node composes cleanly with chat-style graphs. Scalar fields overwrite (LangGraph's default reducer) — each run is the latest consensus.

Note: add_messages is injected into module globals at first use by _state_schema(); the string forward ref here defers resolution.

make_consensus_node

make_consensus_node(swarm: Swarm, *, input_key: str = 'question', sign: bool = False, signing_key: str | None = None, task_id: str | None = None, stream: bool = False) -> Callable[[dict[str, Any]], Awaitable[dict[str, Any]]]

Return an async LangGraph node that runs the swarm and emits a partial state update.

The swarm is captured in the closure — this is the documented way to parameterize a node without breaking the engine, and it keeps the compile-once invariant honest (no per-invoke setup).

which state key holds the prompt. Falls back to the last

human message in state["messages"] if that key is absent — so the node drops into chat-style graphs without configuration.

sign: if True, sign each result and put the ConsensusReceipt (as a dict) into state["receipt"]. The moat, one kwarg away. signing_key: hex Ed25519 private key. If None and sign=True, a fresh keypair is generated ONCE at node construction and reused across invokes — so all receipts from this node are verifiable as coming from the same source. Per-invoke key generation would unlink them. task_id: stable id for the receipt. If None, a uuid4 is generated per invoke (each invoke is a distinct consensus event). stream: if True, forward each per-agent AgentCompleted event (and the final ConsensusResolved) to LangGraph's custom stream channel via get_stream_writer(), so consumers of graph.astream(..., stream_mode="custom") see agents answer in real time. Off = the node is a black box until it returns its state update.

build_consensus_graph

build_consensus_graph(swarm: Swarm, *, input_key: str = 'question', sign: bool = False, signing_key: str | None = None, task_id: str | None = None, stream: bool = False)

Build and compile a standalone consensus subgraph (one node, one edge).

Returns a CompiledStateGraph. Invoke with await graph.ainvoke({"question": ...}), or embed via parent.add_node("consensus", build_consensus_graph(swarm)).

pass True to forward per-agent events to LangGraph's custom stream

channel (consumable via graph.astream(..., stream_mode="custom")).

Compilation happens once here, not per invoke.

claudeway.tools — tool layer

Tool ABCs and the Claudeway/MCP tool wrappers an agent can call. MCP tools are imported lazily (Nostr-style) so the core stays lean.

base

Base Tool class for Claudeway agents.

Tools allow agents to interact with external systems and capabilities.

Tool

Bases: ABC

Base class for all tools that agents can use.

Tools extend agent capabilities by providing structured interfaces to external systems, APIs, or platform features.

execute abstractmethod async

execute(**kwargs) -> dict[str, Any]

Execute the tool with the given parameters.

Returns:

Type Description
dict[str, Any]

A dictionary with the tool execution result.

dict[str, Any]

Should include "success" boolean and either "result" or "error".

get_schema

get_schema() -> dict[str, Any]

Get the tool schema for use in Claude API calls.

Returns a tool definition compatible with Anthropic's tool use format.

format_for_claude

format_for_claude() -> str

Format this tool for inclusion in a Claude system prompt.

Returns a description of what this tool does and how to use it.

claudeway

Claudeway Tool - Allows Claude agents to deploy and manage swarms

This tool enables meta-cognition: Claude can decide when to spawn specialized sub-agents for parallel processing.

ClaudewayTool

ClaudewayTool(runtime)

Bases: Tool

Tool that allows Claude agents to deploy and manage agent swarms.

This enables Claude to become a swarm manager - it can recognize when a task would benefit from multiple specialized agents and deploy them automatically.

Initialize with reference to the runtime.

execute async

execute(**kwargs) -> dict[str, Any]

Execute a Claudeway action.

create_claudeway_tool

create_claudeway_tool(runtime) -> ClaudewayTool

Factory function to create a Claudeway tool.

mcp

MCP tool adapter - let agents call Model Context Protocol servers.

MCP is the 2026 standard for giving agents tools (Goose, Claude Desktop, and others all speak it). This adapter lets you hand an MCP server's tools to a Claudeway Agent exactly like a Python Tool — same execute()/get_schema() shape.

The mcp package is an optional dependency. Importing this module without it installed raises a clear ImportError pointing at pip install claudeway[mcp].

MCPTool

MCPTool(session: Any, tool_name: str, description: str, input_schema: dict[str, Any])

Bases: Tool

Wrap a single MCP server tool as a Claudeway Tool.

The server connection is owned by the caller (an MCPClient session) and passed in; this class only adapts one tool's schema + invocation.

execute async

execute(**kwargs) -> dict[str, Any]

Call the MCP server tool and normalize the result.

MCPClient

MCPClient(transport: Any)

Lazy connection holder for an MCP server.

Usage

async with MCPClient(transport) as mc: tools = await mc.tools() # list[MCPTool] for every server tool agent = Agent(AgentConfig(..., tools=tools))

Kepts the transport open for the agent's lifetime so repeated tool calls don't reconnect each time.

tools async

tools() -> list[MCPTool]

List the server's tools as Claudeway Tool instances.