Getting Started¶

Model your domain¶

Events — facts about what happened.
Commands — intents for what should happen.

Group both under a Namespace; for simple cases, colocate the handler inline. A Command's nested DomainEvents auto-form its return contract (Command.Outcomes):

from langgraph_events import Command, DomainEvent, EventGraph, Namespace


class Order(Namespace):
    class Place(Command):
        customer_id: str
        items: tuple[str, ...]

        class Placed(DomainEvent):
            order_id: str

        class Rejected(DomainEvent):
            reason: str

        def handle(self) -> Placed | Rejected:
            if not self.items:
                return Order.Place.Rejected(reason="empty order")
            return Order.Place.Placed(order_id=f"o-{self.customer_id}")

    class Shipped(DomainEvent):
        tracking: str


graph = EventGraph([Order.Place])
log = graph.invoke(Order.Place(customer_id="alice", items=("book",)))
print(log.latest(Order.Place.Placed))

See Concepts for the full taxonomy and the Outcomes contract. For class-level invariants / raises or cross-domain reactors via @on(...), see Control Flow.

Run the graph¶

log = graph.invoke(seed)                    # sync; returns EventLog
log = await graph.ainvoke(seed)             # async
for event in graph.stream_events(seed): ... # stream as produced

Inspect¶

print(graph.namespaces().text())             # human-readable tree
print(graph.namespaces().mermaid())          # Mermaid diagram
graph.namespaces().namespaces                # structured NamespaceModel access
log.filter(Order.Place.Placed)
log.latest(Order.Place.Rejected)
log.has(Order.Shipped)

Cross-cutting events¶

IntegrationEvent for facts that don't belong to any domain (external signals, shared events). Must live at module scope.

from langgraph_events import Auditable, IntegrationEvent

class MessageReceived(IntegrationEvent):
    text: str

class TaskStarted(IntegrationEvent, Auditable):  # @on(Auditable) for auto-logging
    name: str

Common Tasks¶

I want to...	Reach for...	Docs
Query past events in a handler	`EventLog` (`log.filter()`, `log.latest()`)	Concepts
Enforce a precondition before a handler runs	`invariants` on the `Command` (or `@on()` for non-Command handlers)	Control Flow
Register every inline handler on a domain	`EventGraph.from_namespaces(Order)`	Concepts
Accumulate state across events	`ScalarReducer` on the domain class	Reducers
Accumulate LangChain messages	`message_reducer()`	Reducers
Fan out parallel work	`Scatter[Item]`	Control Flow
Pause for human approval	`Interrupted` + `graph.resume()`	Control Flow
Stop the graph early	Return a `Halted` subclass	Concepts
Catch handler exceptions	`raises` on the `Command` + `@on(HandlerRaised, ...)`	Control Flow
Stream LLM tokens	`astream_events(include_llm_tokens=True)`	Streaming
Connect to an AG-UI frontend	`AGUIAdapter`	AG-UI
Keep old checkpoints working after a rename	`@migrate_from` / `@on(previously=)` + coverage gates	Event Migrations
Know what's safe to change on a live graph	—	Checkpointer Evolution

Where to go next¶

Modelling your domain (taxonomy, handlers, signature injection) → Concepts
Enforcing rules / handling failures (invariants, HITL, exceptions, scatter) → Control Flow
Building a real example → Patterns
Need the full export list? → API Reference