diff --git a/README.md b/README.md
index 025b6b8..8b9b6a2 100644
--- a/README.md
+++ b/README.md
@@ -40,6 +40,26 @@ async def main():
 asyncio.run(main())
 ```
 
+## Code-first workflow DSL
+
+```python
+from orch8 import workflow
+
+checkout = (
+    workflow("checkout")
+    .step("charge", "charge", {"customer_id": "cus_123", "cents": 2500})
+    .parallel(
+        "notify",
+        lambda branch: branch.step("email", "send-email", {"template": "receipt"}),
+        lambda branch: branch.step("audit", "write-audit", {}),
+    )
+    .build()
+)
+```
+
+The builder covers all eleven block types and emits the same JSON accepted by
+`create_sequence`; use ordinary `TypedDict` values for handler-specific params.
+
 ```python
 engine_info = await client.request("GET", "/info")
 ```
diff --git a/src/orch8/__init__.py b/src/orch8/__init__.py
index b5826c5..e12228f 100644
--- a/src/orch8/__init__.py
+++ b/src/orch8/__init__.py
@@ -2,6 +2,8 @@
 from importlib.metadata import version as _version
 
 from .client import Orch8Client
+from .builder import WorkflowBuilder, workflow
+from .adapters import durable_agent_handler
 from .errors import Orch8Error
 from .types import (
     AddResourceRequest,
@@ -113,6 +115,9 @@
     "Orch8Client",
     "Orch8Error",
     "Orch8Worker",
+    "WorkflowBuilder",
+    "workflow",
+    "durable_agent_handler",
     "PluginDef",
     "PoolResource",
     "RegisterDeviceRequest",
diff --git a/src/orch8/adapters.py b/src/orch8/adapters.py
new file mode 100644
index 0000000..0acf831
--- /dev/null
+++ b/src/orch8/adapters.py
@@ -0,0 +1,28 @@
+"""Framework-neutral durable worker adapters."""
+
+from __future__ import annotations
+
+import inspect
+from collections.abc import Callable
+from typing import Any
+
+
+def durable_agent_handler(runner: Any) -> Callable[[Any], Any]:
+    """Adapt LangGraph/CrewAI/AutoGen-style runners to an Orch8 handler."""
+
+    async def handle(task: Any) -> Any:
+        params = task.params if hasattr(task, "params") else task["params"]
+        instance_id = getattr(task, "instance_id", None) or (
+            task.get("instance_id") if isinstance(task, dict) else None
+        )
+        config = {"configurable": {"thread_id": instance_id}} if instance_id else None
+        if callable(getattr(runner, "ainvoke", None)):
+            return await runner.ainvoke(params, config)
+        for name in ("invoke", "kickoff", "run"):
+            method = getattr(runner, name, None)
+            if callable(method):
+                result = method(params, config) if name == "invoke" else method(params)
+                return await result if inspect.isawaitable(result) else result
+        raise TypeError("agent runner must implement ainvoke, invoke, kickoff, or run")
+
+    return handle
diff --git a/src/orch8/builder.py b/src/orch8/builder.py
new file mode 100644
index 0000000..7a3b657
--- /dev/null
+++ b/src/orch8/builder.py
@@ -0,0 +1,178 @@
+"""Fluent, dependency-free builder for the Orch8 workflow JSON DSL."""
+
+from __future__ import annotations
+
+from collections.abc import Callable, Mapping, Sequence
+from typing import Any
+
+Block = dict[str, Any]
+Branch = Callable[["WorkflowBuilder"], None]
+
+
+class WorkflowBuilder:
+    def __init__(self, name: str, namespace: str = "default") -> None:
+        if not name:
+            raise ValueError("workflow name cannot be empty")
+        self.name = name
+        self.namespace = namespace
+        self._items: list[Block] = []
+
+    def step(
+        self,
+        id: str,
+        handler: str,
+        params: Mapping[str, Any] | None = None,
+        **options: Any,
+    ) -> WorkflowBuilder:
+        self._items.append(
+            {"type": "step", "id": id, "handler": handler, "params": dict(params or {}), **options}
+        )
+        return self
+
+    def parallel(self, id: str, *branches: Branch) -> WorkflowBuilder:
+        self._items.append(
+            {"type": "parallel", "id": id, "branches": [self._branch(fn) for fn in branches]}
+        )
+        return self
+
+    def race(self, id: str, *branches: Branch, semantics: str | None = None) -> WorkflowBuilder:
+        block: Block = {"type": "race", "id": id, "branches": [self._branch(fn) for fn in branches]}
+        if semantics is not None:
+            block["semantics"] = semantics
+        self._items.append(block)
+        return self
+
+    def loop(
+        self, id: str, condition: str, body: Branch, *, max_iterations: int = 1000, **options: Any
+    ) -> WorkflowBuilder:
+        self._items.append(
+            {
+                "type": "loop",
+                "id": id,
+                "condition": condition,
+                "body": self._branch(body),
+                "max_iterations": max_iterations,
+                **options,
+            }
+        )
+        return self
+
+    def for_each(
+        self, id: str, collection: str, body: Branch, *, item_var: str = "item", **options: Any
+    ) -> WorkflowBuilder:
+        self._items.append(
+            {
+                "type": "for_each",
+                "id": id,
+                "collection": collection,
+                "item_var": item_var,
+                "body": self._branch(body),
+                **options,
+            }
+        )
+        return self
+
+    def router(
+        self,
+        id: str,
+        routes: Sequence[tuple[str, Branch]],
+        *,
+        default: Branch | None = None,
+    ) -> WorkflowBuilder:
+        block: Block = {
+            "type": "router",
+            "id": id,
+            "routes": [
+                {"condition": condition, "blocks": self._branch(branch)}
+                for condition, branch in routes
+            ],
+        }
+        if default is not None:
+            block["default"] = self._branch(default)
+        self._items.append(block)
+        return self
+
+    def try_catch(
+        self,
+        id: str,
+        try_block: Branch,
+        catch_block: Branch,
+        *,
+        finally_block: Branch | None = None,
+    ) -> WorkflowBuilder:
+        block: Block = {
+            "type": "try_catch",
+            "id": id,
+            "try_block": self._branch(try_block),
+            "catch_block": self._branch(catch_block),
+        }
+        if finally_block is not None:
+            block["finally_block"] = self._branch(finally_block)
+        self._items.append(block)
+        return self
+
+    def sub_sequence(
+        self, id: str, sequence_name: str, *, version: int | None = None, input: Any = None
+    ) -> WorkflowBuilder:
+        block: Block = {"type": "sub_sequence", "id": id, "sequence_name": sequence_name}
+        if version is not None:
+            block["version"] = version
+        if input is not None:
+            block["input"] = input
+        self._items.append(block)
+        return self
+
+    def ab_split(
+        self, id: str, variants: Sequence[tuple[str, int, Branch]]
+    ) -> WorkflowBuilder:
+        self._items.append(
+            {
+                "type": "ab_split",
+                "id": id,
+                "variants": [
+                    {"name": name, "weight": weight, "blocks": self._branch(branch)}
+                    for name, weight, branch in variants
+                ],
+            }
+        )
+        return self
+
+    def cancellation_scope(self, id: str, body: Branch) -> WorkflowBuilder:
+        self._items.append(
+            {"type": "cancellation_scope", "id": id, "blocks": self._branch(body)}
+        )
+        return self
+
+    def saga(
+        self,
+        id: str,
+        steps: Sequence[tuple[str, Branch, Branch | None]],
+    ) -> WorkflowBuilder:
+        resolved = []
+        for step_id, action, compensation in steps:
+            actions = self._branch(action)
+            compensations = self._branch(compensation) if compensation else []
+            if len(actions) != 1 or len(compensations) > 1:
+                raise ValueError("each saga action must have one block and compensation at most one")
+            item: Block = {"id": step_id, "action": actions[0]}
+            if compensations:
+                item["compensation"] = compensations[0]
+            resolved.append(item)
+        self._items.append({"type": "saga", "id": id, "steps": resolved})
+        return self
+
+    def raw(self, block: Mapping[str, Any]) -> WorkflowBuilder:
+        self._items.append(dict(block))
+        return self
+
+    def build(self) -> Block:
+        return {"name": self.name, "namespace": self.namespace, "blocks": list(self._items)}
+
+    def _branch(self, callback: Branch) -> list[Block]:
+        inner = WorkflowBuilder("_inner", self.namespace)
+        callback(inner)
+        return inner._items
+
+
+def workflow(name: str, namespace: str = "default") -> WorkflowBuilder:
+    return WorkflowBuilder(name, namespace)
diff --git a/src/orch8/types.py b/src/orch8/types.py
index dfb90d4..a68db6c 100644
--- a/src/orch8/types.py
+++ b/src/orch8/types.py
@@ -55,6 +55,8 @@ class ExecutionContext(BaseModel):
 
 
 class SequenceDefinition(BaseModel):
+    schema_url: str | None = Field(default=None, alias="$schema")
+    schema_version: int = 1
     id: str
     tenant_id: str
     namespace: str
diff --git a/tests/test_adapters.py b/tests/test_adapters.py
new file mode 100644
index 0000000..785b34a
--- /dev/null
+++ b/tests/test_adapters.py
@@ -0,0 +1,15 @@
+import asyncio
+
+from orch8 import durable_agent_handler
+
+
+class Graph:
+    async def ainvoke(self, value, config):
+        return value, config
+
+
+def test_langgraph_thread_is_instance_id() -> None:
+    result = asyncio.run(
+        durable_agent_handler(Graph())({"params": {"text": "hi"}, "instance_id": "inst-1"})
+    )
+    assert result == ({"text": "hi"}, {"configurable": {"thread_id": "inst-1"}})
diff --git a/tests/test_builder.py b/tests/test_builder.py
new file mode 100644
index 0000000..147deb8
--- /dev/null
+++ b/tests/test_builder.py
@@ -0,0 +1,34 @@
+from orch8 import workflow
+
+
+def test_builder_covers_nested_and_ab_split_blocks() -> None:
+    definition = (
+        workflow("campaign")
+        .step("prepare", "prepare", {"audience": "new"})
+        .parallel(
+            "fanout",
+            lambda branch: branch.step("email", "send-email", {"template": "welcome"}),
+            lambda branch: branch.ab_split(
+                "copy",
+                [
+                    ("a", 50, lambda variant: variant.step("a", "render", {"copy": "A"})),
+                    ("b", 50, lambda variant: variant.step("b", "render", {"copy": "B"})),
+                ],
+            ),
+        )
+        .build()
+    )
+    assert definition["name"] == "campaign"
+    assert definition["blocks"][1]["branches"][1][0]["type"] == "ab_split"
+
+
+def test_saga_rejects_multiple_action_blocks() -> None:
+    try:
+        workflow("bad").saga(
+            "saga",
+            [("one", lambda branch: branch.step("a", "noop").step("b", "noop"), None)],
+        )
+    except ValueError as error:
+        assert "one block" in str(error)
+    else:
+        raise AssertionError("invalid saga was accepted")