GraphBus Documentation

# Clone the repo
git clone https://github.com/graphbus/graphbus-core
cd graphbus-core

# Install (editable mode for development)
pip install -e .

# Verify
graphbus --version
# graphbus 0.1.0

# Claude (recommended)
export ANTHROPIC_API_KEY=sk-ant-...

# GraphBus works without an API key in static build mode

graphbus init hello-graphbus --intent "generate and log greeting messages"
cd hello-graphbus

from graphbus_core import GraphBusNode, schema_method, subscribe

class HelloService(GraphBusNode):
    """Generates and logs greeting messages."""

    SYSTEM_PROMPT = """
    I generate friendly greeting messages.
    During build cycles, I negotiate with other agents to ensure
    my output schema is consistent and well-typed.
    """

    @schema_method(
        input_schema={},
        output_schema={"message": str, "timestamp": str}
    )
    def generate_message(self) -> dict:
        import datetime
        return {
            "message": "Hello from GraphBus!",
            "timestamp": datetime.datetime.utcnow().isoformat()
        }

    @subscribe("/Hello/MessageGenerated")
    def on_message(self, event):
        self.log(f"Received: {event.payload['message']}")

graphbus build agents/

# [BUILD] Scanning agents/hello_service.py
# [BUILD] Discovered: HelloService
# [BUILD] Graph: 1 node, 0 edges
# [BUILD] Schema validated: generate_message → {message: str, timestamp: str}
# [BUILD] Artifacts written to .graphbus/
# [BUILD] Complete in 0.12s

graphbus run .graphbus/

# [RUNTIME] Loaded 1 agent (HelloService)
# [RUNTIME] Message bus initialized — 1 topic
# [RUNTIME] HelloService → "Hello from GraphBus!"
# [RUNTIME] Zero LLM calls made.

export ANTHROPIC_API_KEY=sk-ant-...
graphbus negotiate .graphbus --intent "add error handling for clock failures"

# ✦ Intent: "add error handling for clock failures"
# [AGENT] HelloService: "I propose wrapping datetime.utcnow() in a try/except"
# [ARBITER] Accepted (no other agents to consult)
# ✓ 1 file improved (agents/hello_service.py)
# ✓ Artifacts written to .graphbus/

my-project/
├── agents/
│   ├── order_processor.py      ← GraphBusNode subclasses
│   ├── inventory_service.py
│   └── notification_service.py
├── .graphbus/                  ← Build artifacts (auto-generated)
│   ├── graph.json              ← Agent dependency graph
│   ├── agents.json             ← Agent metadata + source
│   ├── topics.json             ← Topic registry
│   └── build_summary.json      ← Build metadata
├── graphbus.yaml               ← Project config (optional)
└── tests/

# Set intent at project creation
graphbus init my-service --intent "order processing pipeline with payment validation"

# Reference it in any negotiation
graphbus negotiate .graphbus --intent "add null amount validation to order processor"

from graphbus_core import GraphBusNode

class MyAgent(GraphBusNode):
    # Optional: LLM system prompt for build-mode reasoning
    SYSTEM_PROMPT = "Describe this agent's role and negotiation strategy."

    # Optional: mark as arbiter (resolves conflicts)
    IS_ARBITER = False

    # Optional: custom memory backend
    memory = None

    # Your methods go here...

@schema_method(
    input_schema={"order_id": str, "items": list},
    output_schema={"status": str, "total": float}
)
def process_order(self, order_id: str, items: list) -> dict:
    ...

@subscribe("/Order/Created")
def on_order_created(self, event):
    # event.topic, event.payload, event.timestamp
    self.log(f"New order: {event.payload['order_id']}")

from graphbus_core import depends_on

class NotificationService(GraphBusNode):
    @depends_on("OrderProcessor")
    def send_confirmation(self, order_id: str):
        # OrderProcessor is guaranteed to have run first
        ...

# Publishing (in any @schema_method or handler)
self.publish("/Order/Created", {
    "order_id": "ord_123",
    "total": 49.99
})

# Subscribing (declare on the class)
@subscribe("/Order/Created")
def handle_order(self, event):
    print(event.payload["order_id"])  # ord_123

{
  "nodes": [
    {
      "id": "HelloService",
      "module": "agents.hello_service",
      "system_prompt": "...",
      "methods": ["generate_message"],
      "subscriptions": ["/Hello/MessageGenerated"]
    }
  ],
  "edges": [],
  "topological_order": ["HelloService"]
}

graphbus inspect .graphbus/
# Shows: agent graph, schemas, topic registry, build metadata

graphbus validate agents/
# Validates schemas and dependency edges without building

# Static build first (fast, no LLM)
graphbus build agents/

# Then negotiate with a specific intent
graphbus negotiate .graphbus --intent "improve error handling in order processor"

# Negotiate with a specific model
graphbus negotiate .graphbus --intent "add retry logic" --llm-model deepseek/deepseek-reasoner

class Proposal:
    agent_id: str        # who is proposing
    intent: str          # the user's goal this proposal serves
    target_file: str     # which file to change
    diff: str            # unified diff of proposed changes
    rationale: str       # LLM's reasoning tied back to intent
    affects: list[str]   # other agents whose schemas are impacted

class Evaluation:
    agent_id: str        # who is evaluating
    proposal_id: str     # which proposal
    decision: str        # "accept" | "reject"
    reasoning: str       # LLM's reasoning
    counter_proposal: str | None  # alternative diff if rejecting

graphbus inspect-negotiation
# Browse: proposals, evaluations, arbitration decisions, commits

class ArbiterService(GraphBusNode):
    IS_ARBITER = True  # Mark as the system arbiter

    SYSTEM_PROMPT = """
    You are an impartial arbiter. When agents disagree on a proposal:
    - Favor changes that improve correctness and maintainability
    - Reject changes that introduce risk without clear benefit
    - Be conservative — when in doubt, reject
    - Provide clear reasoning for every decision
    """

# Programmatic runtime usage
from graphbus_core.runtime import RuntimeExecutor, RuntimeConfig

config = RuntimeConfig(artifacts_dir=".graphbus/")
executor = RuntimeExecutor(config)

executor.start()
result = executor.invoke("HelloService", "generate_message", {})
print(result)  # {"message": "Hello from GraphBus!", "timestamp": "..."}
executor.stop()

graphbus run .graphbus/
graphbus run .graphbus/ --watch    # hot reload on artifact changes
graphbus run .graphbus/ --debug    # interactive debugger

# Agent A publishes
self.publish("/Order/Created", {"order_id": "ord_123"})

# Router dispatches to all @subscribe("/Order/Created") handlers
# across all loaded agents — in topological order

# Agent B receives
@subscribe("/Order/Created")
def handle_order(self, event):
    # event.topic   = "/Order/Created"
    # event.payload = {"order_id": "ord_123"}
    # event.timestamp = datetime object

cd examples/hello_graphbus
python build.py                         # static build
ANTHROPIC_API_KEY=sk-... python build.py  # agent build
python run.py                           # run artifacts

cd examples/hello_world_mcp
graphbus build agents/
graphbus run .graphbus/ --mcp
# Exposes agents as MCP tools on ws://localhost:8765

cd examples/news_summarizer
graphbus build agents/
graphbus run .graphbus/