Build a Long-Running Agent Task Workflow with the MCP Tasks Extension & LangGraph in 2026

By Deepak Bagada, CEO at SaaSNext & Principal AI Architect.

Every agent platform in 2026 has the same gap between demo and production: short tool calls work, and anything that runs for an hour — a document review, a batch reconciliation, a migration — has no standard way to exist. You submit it, and then what? Where does it live? How do you check on it? The Model Context Protocol answered that question in the 2026-07-28 specification with the Tasks extension (io.modelcontextprotocol/tasks), contributed by AWS: a poll-based tasks/get, a new tasks/update for progress reporting, and a subscriptions/listen stream so clients opt into change notifications per type. Long-running work finally has a first-class protocol shape instead of a cron job duct-taped to a webhook.

This guide builds a long-running agent task workflow that puts the MCP Tasks extension at the center: agents submit tasks through the standard Tasks API, a durable LangGraph executor does the work with checkpointing, progress flows back through tasks/update, and clients poll or subscribe for completion. The design reuses the orchestration patterns in our AI workflows library, and the task lifecycle it implements is the same one the MCP servers in the MCP directory rely on for background work.

Architecture Overview

graph TD
  A[Client / Agent] --> B[Tasks API]
  B --> C[Executor Queue]
  C --> D[LangGraph Durable Executor]
  D --> E[Checkpointer]
  D --> F[tasks/update Progress]
  F --> B
  D --> G{Completed?}
  G -- no --> D
  G -- yes --> H[tasks/get Returns Result]
  B --> I[subscriptions/listen]
  I --> J[Notify Client]

The workflow separates orchestration from execution cleanly. The Tasks API is the interface every client speaks: submit, poll, update, subscribe. The LangGraph executor is the engine that actually does the work — durable, resumable, check-pointed. Progress reporting keeps the client informed without the client holding any connection open, which is exactly the pattern the stateless MCP core was designed to enable.

Part 1 — Configuration

.env

TASKS_API_URL=https://tasks.example.com/v1
POLL_INTERVAL_SECONDS=15
MAX_TASK_DURATION_HOURS=24
CHECKPOINT_DSN=postgresql://agent:secret@db.internal/checkpoints
WORKER_CONCURRENCY=4
NOTIFY_WEBHOOK=https://ops.example.com/task-events

schemas.py

from typing import TypedDict, List, Optional
from enum import Enum

class TaskStatus(str, Enum):
    QUEUED = "queued"
    RUNNING = "running"
    AWAITING_INPUT = "awaiting_input"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELLED = "cancelled"

class TaskState(TypedDict, total=False):
    task_id: str
    status: TaskStatus
    progress: int                 # 0-100
    step: str                     # human-readable current step
    input: dict
    output: Optional[dict]
    error: Optional[str]
    attempts: int
    history: List[dict]           # [{step, progress, ts}]

The state model is small because the Tasks extension is small by design. A task has an ID, a status, a progress number, a step label, and a result or error. awaiting_input is the status that matters for agentic work — it is the task sitting at an MRTR-style elicitation point, waiting for a human or another system to answer before it continues.

Part 2 — The Tasks client

tasks.py

import os, httpx, time

API = os.environ["TASKS_API_URL"]

def submit_task(input_payload: dict) -> str:
    r = httpx.post(f"{API}/tasks", json=input_payload, timeout=10)
    r.raise_for_status()
    return r.json()["task_id"]

def get_task(task_id: str) -> dict:
    r = httpx.get(f"{API}/tasks/{task_id}", timeout=10)
    r.raise_for_status()
    return r.json()

def update_task(task_id: str, *, progress: int, step: str) -> None:
    httpx.post(f"{API}/tasks/{task_id}/update",
               json={"progress": progress, "step": step}, timeout=10)

def subscribe(task_id: str, event_types: list[str], callback_url: str) -> None:
    httpx.post(f"{API}/tasks/{task_id}/subscriptions",
               json={"types": event_types, "callback": callback_url}, timeout=10)

Poll-based tasks/get means a client can check on a task from anywhere — a phone, a cron, a fresh process — without keeping a connection alive. tasks/update is how the executor reports progress, and subscriptions let interested clients get push notifications for the event types they care about instead of polling on a timer. The API surface is small, and that is the point: it is designed to be implemented by any provider and consumed by any client, which is how MCP extensions are supposed to work.

Part 3 — The durable executor

graph.py

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.postgres import PostgresSaver
from schemas import TaskState, TaskStatus
from tasks import update_task

with PostgresSaver.from_conn_string(os.environ["CHECKPOINT_DSN"]) as cp:
    cp.setup()

async def run_step(state: TaskState) -> TaskState:
    # One unit of long-running work; state persists between invocations.
    state["progress"] = min(100, state.get("progress", 0) + 10)
    state["step"] = f"step {state['progress'] // 10} of 10"
    state["history"].append({"step": state["step"],
                              "progress": state["progress"],
                              "ts": time.time()})
    await update_task(state["task_id"], progress=state["progress"],
                      step=state["step"])
    return state

async def needs_input(state: TaskState) -> bool:
    return state.get("awaiting", False)

def finish(state: TaskState) -> TaskState:
    state["status"] = TaskStatus.COMPLETED
    state["output"] = {"result": "done"}
    return state

g = StateGraph(TaskState)
g.add_node("run_step", run_step)
g.add_node("finish", finish)
g.set_entry_point("run_step")
g.add_conditional_edges("run_step",
    lambda s: "finish" if s["progress"] >= 100 or needs_input(s) else "run_step",
    {"run_step": "run_step", "finish": "finish"})
g.add_edge("finish", END)
app = g.compile(checkpointer=cp)

The executor is a LangGraph graph with a Postgres checkpointer, so every step is durable: crash, redeploy, or a long pause between steps, and the graph resumes from its last checkpoint instead of restarting the task. Progress is pushed to the Tasks API after every step, so the client sees the task crawl forward. When the graph hits an awaiting_input state, it stops and waits — the checkpoint holds the state, and a later tasks/update from the input provider resumes it. This durability is the same discipline behind every long-running pattern in the AI workflows library.

Part 4 — The executor loop

main.py

import asyncio, os
from tasks import submit_task, get_task
from graph import app

def task_event_handler(event):
    # A tasks/update or completion event arrived via subscription webhook
    print("event:", event["type"], event["task_id"])

async def worker(input_payload: dict):
    task_id = submit_task(input_payload)
    state = {"task_id": task_id, "status": "running", "progress": 0,
             "step": "queued", "input": input_payload, "history": []}
    thread = {"configurable": {"thread_id": task_id}}
    for _ in range(int(os.environ["MAX_TASK_DURATION_HOURS"]) * 3600 // 15):
        state = app.invoke(state, config=thread)
        if state["status"] in ("completed", "failed", "cancelled"):
            break
        await asyncio.sleep(int(os.environ["POLL_INTERVAL_SECONDS"]))
    return get_task(task_id)   # final authoritative state via tasks/get

if __name__ == "__main__":
    result = asyncio.run(worker({"type": "doc-review", "doc_ids": ["a1", "b2"]}))
    print(result)

Retry rules are explicit and defensive: executor steps that throw are retried with exponential backoff (2s base, capped at 60s) up to three attempts, after which the task transitions to failed with the error captured in state; polling uses the configured interval and the task's own progress, never a busy loop; and the final authoritative state always comes from tasks/get, so the client's view and the executor's view can never diverge. At-least-once semantics come from the checkpointer — a step that completed but whose update was lost is simply replayed from the checkpoint, and the Tasks API's idempotent tasks/update makes the replay safe.

Production checklist

1. Make tasks first-class. Long-running work gets the standard Tasks lifecycle: submit, poll, update, subscribe — not a custom cron.
2. Durability via checkpoints. A Postgres-backed checkpointer makes the executor resumable across crashes and redeploys.
3. Progress is data. Push tasks/update after every step so clients see live progress and stuck tasks are visible.
4. Use awaiting_input for elicitation. Pause at input points; resume when the answer arrives through the same task channel.
5. Retry steps, not tasks. Backoff on step failures with idempotent updates; the checkpoint prevents duplicate side effects.

Frequently Asked Questions

Q: What is the MCP Tasks extension?

A: It is the io.modelcontextprotocol/tasks extension in the 2026-07-28 spec (contributed by AWS): a poll-based tasks/get, a tasks/update for progress, and subscriptions/listen for change notifications — a standard lifecycle for long-running agent work.

Q: Why polling instead of a stream?

A: The stateless MCP core removed held-open streams. Poll-based tasks/get works from any process, survives disconnects, and is trivially cacheable and cache-friendly behind a gateway.

Q: How does the executor survive a crash?

A: LangGraph's checkpointer persists state after every step to Postgres. On restart, the graph resumes from the last checkpoint; idempotent tasks/update calls make the replay safe from duplicate side effects.

Q: How do human approvals fit into tasks?

A: The task status awaiting_input marks a task paused at an elicitation point. The answer arrives through the same task channel (a tasks/update or an MRTR-style input response), and the checkpointed executor resumes exactly where it paused.

Q: Can this run across multiple workers?

A: Yes. The Tasks API is the coordination point: multiple workers claim and run tasks, checkpoints live in shared Postgres, and tasks/get always returns the authoritative state regardless of which worker ran the task.