Skip to content

API Reference

This page documents the public API exported from azure_functions_durable_graph.

from azure_functions_durable_graph import (
    DurableGraphApp,
    GraphManifest,
    GraphRegistration,
    ManifestBuilder,
    RouteAction,
    RouteDecision,
)

Public surface

The package exports DurableGraphApp, GraphManifest, GraphRegistration, ManifestBuilder, RouteAction, and RouteDecision. Registry and contract internals are not public contracts.

ManifestBuilder

Bases: Generic[StateModelT]

Source code in src/azure_functions_durable_graph/manifest.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def __init__(
    self,
    *,
    graph_name: str,
    state_model: type[StateModelT],
    version: str = "0.1.0",
    metadata: dict[str, Any] | None = None,
) -> None:
    self.graph_name = graph_name
    self.state_model = state_model
    self.version = version
    self.metadata = metadata or {}
    self.entrypoint: str | None = None
    self._nodes: dict[str, NodeDefinition] = {}
    self._node_handlers: dict[str, NodeHandler[StateModelT]] = {}
    self._route_handlers: dict[str, RouteHandler[StateModelT]] = {}
    self._event_handlers: dict[str, EventHandler[StateModelT]] = {}

Usage example: building a graph

from pydantic import BaseModel

from azure_functions_durable_graph import ManifestBuilder, RouteDecision


class AgentState(BaseModel):
    query: str
    classified: bool = False
    response: str | None = None


def classify(state: AgentState) -> dict:
    return {"classified": True}


def route(state: AgentState) -> RouteDecision:
    return RouteDecision.next("respond")


def respond(state: AgentState) -> dict:
    return {"response": f"Answer to: {state.query}"}


builder = ManifestBuilder(graph_name="agent", state_model=AgentState)
builder.set_entrypoint("classify")
builder.add_node("classify", classify, route=route)
builder.add_node("respond", respond, terminal=True)
registration = builder.build()

GraphManifest

Bases: BaseModel

The manifest is a Pydantic model containing:

  • graph_name — unique graph identifier
  • version — semantic version string
  • graph_hash — SHA-256 hash of canonical topology JSON (first 16 chars)
  • state_model_name — name of the Pydantic state model class
  • entrypoint — starting node name
  • nodes — dict of NodeDefinition objects
  • event_handler_names — sorted list of registered event handler names
  • metadata — arbitrary metadata dict

GraphRegistration

Bases: Generic[StateModelT]

A dataclass containing:

  • manifest — the compiled GraphManifest
  • state_model — the Pydantic state model type
  • node_handlers — dict mapping handler names to callables
  • route_handlers — dict mapping route handler names to callables
  • event_handlers — dict mapping event names to callables

RouteDecision

Bases: BaseModel

Factory methods

# Continue to next node
RouteDecision.next("step_two")

# Complete the graph
RouteDecision.complete(note="all done")

# Wait for an external event
RouteDecision.wait_for_event(
    event_name="approval",
    resume_node="process",
)

RouteAction

Bases: str, Enum

An enum with values:

  • RouteAction.NEXT — continue to next node
  • RouteAction.COMPLETE — finish the graph
  • RouteAction.WAIT_FOR_EVENT — pause for external event

DurableGraphApp

Source code in src/azure_functions_durable_graph/app.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def __init__(
    self,
    *,
    auth_level: func.AuthLevel = func.AuthLevel.ANONYMOUS,
) -> None:
    self.function_app = func.FunctionApp(http_auth_level=auth_level)
    self.blueprint = df.Blueprint()
    self.registry = GraphRegistry()

    self._orchestrator_name = "afdg_orchestrator"
    self._node_activity_name = "afdg_execute_node"
    self._route_activity_name = "afdg_resolve_route"
    self._event_activity_name = "afdg_apply_event"

    self._register_runtime_functions()
    self.function_app.register_functions(self.blueprint)

Import safety

DurableGraphApp requires azure-functions and azure-functions-durable at import time. In pure unit test environments without these packages, DurableGraphApp is set to None in __init__.py.

Usage example: wiring a Function App

from azure_functions_durable_graph import DurableGraphApp

runtime = DurableGraphApp()
runtime.register_registration(registration)
app = runtime.function_app

HTTP API endpoints

DurableGraphApp registers the following endpoints:

Method Route Description
POST /api/graphs/{graph_name}/runs Start a new graph run
GET /api/runs/{instance_id} Get run status
POST /api/runs/{instance_id}/events/{event_name} Send an external event
POST /api/runs/{instance_id}/cancel Cancel a run
GET /api/openapi.json OpenAPI document
GET /api/health Health check with registered graphs

Internal references

These modules are useful for advanced extension work but are internal APIs:

  • registry.py: GraphRegistry, execute_node, resolve_route, apply_event
  • contracts.py: OrchestrationInput, NodeExecutionRequest, RouteResolutionRequest, EventApplyRequest, RunStatusEnvelope

For full implementation patterns, see Usage and Architecture.