dag-planner-mcp

A durable DAG-based task planner exposed as an MCP (Model Context Protocol) server. It lets AI orchestrators (Claude, ADK-based agents, etc.) break a goal into a dependency graph of tasks, execute them in parallel where possible, track state durably, and handle human-in-the-loop approval — all through a clean set of 22 MCP tools.

Table of Contents

• Quick start with uvx
• Install this skill
• Requirements
• Installation
• Database Configuration
  • SQLite (development)
  • PostgreSQL (production)
• Environment Variables
• Running the server
• Using with Claude Desktop (stdio)
• Using with other MCP clients (streamable HTTP)
• Streamlit Dashboard
• Available MCP Tools
• Orchestrator Loop Example
• Running Tests

Quick start with uvx

The fastest way to run the server is with uvx — no virtual environment or pip install needed:

uvx dag-planner-mcp

Pass arguments (e.g. HTTP transport) the same way:

uvx dag-planner-mcp --transport streamable-http --host 0.0.0.0 --port 8000

With an environment variable:

DATABASE_URL="sqlite:///dag_planner.db" uvx dag-planner-mcp

Claude Desktop — one-line config (uvx)

Open claude_desktop_config.json and add:

{
  "mcpServers": {
    "dag-planner-mcp": {
      "command": "uvx",
      "args": ["dag-planner-mcp"],
      "env": {
        "DATABASE_URL": "sqlite:////home/user/data/dag_planner.db"
      }
    }
  }
}

No installation step is required — uvx fetches and caches the package automatically on first run.

Install this skill

AI agents (Claude, Copilot, etc.) can pick up ready-made instructions for using this MCP server by installing the bundled skill:

npx skills add Shubhamnegi/dag-planner-mcp --skill use-mcp-tool

Or install directly from the skill path:

npx skills add https://github.com/Shubhamnegi/dag-planner-mcp/tree/main/skills/use-mcp-tool

What the skill provides

Requirements

• Python ≥ 3.11
• DATABASE_URL environment variable (defaults to sqlite:///dag_planner.db)

Requirements

Optional (PostgreSQL):

Optional (Dashboard):

Installation

1. Clone the repository

git clone https://github.com/Shubhamnegi/dag-planner-mcp.git
cd dag-planner-mcp

2. Create and activate a virtual environment

python -m venv .venv
source .venv/bin/activate        # Linux / macOS
.venv\Scripts\activate           # Windows

3. Install the package

SQLite (dev — no extra dependencies):

pip install -e .

PostgreSQL (prod):

pip install -e ".[postgres]"

With development/test dependencies:

pip install -e ".[dev]"

With Streamlit dashboard:

pip install -e ".[dashboard]"

Database Configuration

The server is controlled entirely via the DATABASE_URL environment variable. Tables are created automatically on first start.

SQLite (development)

# Default — creates dag_planner.db in the current directory
export DATABASE_URL="sqlite:///dag_planner.db"

# Absolute path
export DATABASE_URL="sqlite:////home/user/data/dag_planner.db"

# In-memory (testing only — data lost on exit)
export DATABASE_URL="sqlite:///:memory:"

No additional setup is required for SQLite.

PostgreSQL (production)

export DATABASE_URL="postgresql://user:password@localhost:5432/dag_planner"

Create the database first:

CREATE DATABASE dag_planner;

Then start the server — SQLAlchemy will create all tables automatically.

For connection pooling / SSL in production you can pass extra query parameters:

export DATABASE_URL="postgresql://user:password@host:5432/dag_planner?sslmode=require"

Environment Variables

Running the server

stdio (recommended for Claude Desktop and most MCP clients)

dag-planner-mcp
# or
python -m dag_planner_mcp.server

The server reads from stdin and writes to stdout — no port is opened.

Streamable HTTP

dag-planner-mcp --transport streamable-http --host 0.0.0.0 --port 8000

The MCP endpoint will be available at:

http://localhost:8000/mcp

Using with Claude Desktop (stdio)

The recommended approach is to use uvx so no manual installation is needed (see Quick start with uvx above).

If you prefer to point at a locally installed binary:

1. Open the Claude Desktop configuration file:

   • macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
   • Windows: %APPDATA%\Claude\claude_desktop_config.json

2. Add the server under mcpServers:

{
  "mcpServers": {
    "dag-planner-mcp": {
      "command": "/path/to/.venv/bin/dag-planner-mcp",
      "env": {
        "DATABASE_URL": "sqlite:////home/user/data/dag_planner.db"
      }
    }
  }
}

Replace /path/to/.venv/bin/dag-planner-mcp with the absolute path to the installed script (run which dag-planner-mcp after installation).

3. Restart Claude Desktop. The 22 DAG planner tools will appear in the tools panel.

Using with other MCP clients (streamable HTTP)

Start the server in HTTP mode:

DATABASE_URL="sqlite:///dag_planner.db" \
dag-planner-mcp --transport streamable-http --host 0.0.0.0 --port 8000

Then point your MCP client at:

http://localhost:8000/mcp

Example: Cursor IDE

{
  "mcpServers": {
    "dag-planner-mcp": {
      "url": "http://localhost:8000/mcp"
    }
  }
}

Example: Windsurf / Continue / custom agent

from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

async with streamablehttp_client("http://localhost:8000/mcp") as (r, w, _):
    async with ClientSession(r, w) as session:
        await session.initialize()
        result = await session.call_tool("create_workflow_run", {"goal": "Analyze AWS costs"})

Streamlit Dashboard

A read-only Streamlit dashboard ships in the dashboard/ directory. It reads directly from the same database as the MCP server (via DATABASE_URL) and never writes any data.

Pages

Quick start

# Install with dashboard extras
pip install -e ".[dashboard]"

# Point at the same database your MCP server uses
export DATABASE_URL="sqlite:///dag_planner.db"
# — or for PostgreSQL —
export DATABASE_URL="postgresql://user:password@localhost:5432/dag_planner"

# Launch
streamlit run dashboard/app.py

The dashboard opens at http://localhost:8501 by default.

DAG visualization requires the graphviz system package in addition to the Python bindings. Install it with brew install graphviz (macOS) or apt-get install graphviz (Debian/Ubuntu). If the system package is absent the page falls back to a plain adjacency table.

Available MCP Tools

Planning

Scheduling

State Management

Task I/O

Query

Validation

Subagent-safe wrappers

Orchestrator Loop Example

import json
from mcp import ClientSession
from mcp.client.stdio import stdio_client

async def run():
    async with stdio_client(["dag-planner-mcp"]) as (r, w):
        async with ClientSession(r, w) as session:
            await session.initialize()

            # 1. Create a workflow
            res = await session.call_tool("create_workflow_run", {
                "goal": "Analyze AWS cost spike and send report"
            })
            run_id = json.loads(res.content[0].text)["data"]["run_id"]

            # 2. Define the task DAG
            await session.call_tool("create_plan_graph", {
                "run_id": run_id,
                "tasks": [
                    {
                        "task_key": "fetch_data",
                        "title": "Fetch cost data",
                        "description": "Pull last 3 weeks of AWS cost data",
                        "owner_agent": "data_agent",
                        "depends_on": [],
                        "output_contract": {"type": "object", "required": ["cost_data"]}
                    },
                    {
                        "task_key": "analyze",
                        "title": "Analyze spike",
                        "description": "Identify top services causing the spike",
                        "owner_agent": "analyst_agent",
                        "depends_on": ["fetch_data"],
                        "output_contract": {"type": "object", "required": ["summary"]}
                    }
                ]
            })

            # 3. Execution loop
            while True:
                res = await session.call_tool("get_ready_tasks", {"run_id": run_id})
                tasks = json.loads(res.content[0].text)["data"]["tasks"]

                if not tasks:
                    break  # All done (or blocked)

                for task in tasks:
                    task_id = task["task_id"]
                    await session.call_tool("claim_task_for_execution", {
                        "task_id": task_id, "executor_id": "orchestrator-1"
                    })
                    await session.call_tool("mark_task_running", {"task_id": task_id})

                    # ... dispatch to subagent, collect result ...
                    output = {"summary": "EC2 caused 40% increase"}

                    await session.call_tool("put_task_output", {
                        "task_id": task_id, "output": output, "is_final": True
                    })
                    await session.call_tool("validate_task_output", {"task_id": task_id})
                    await session.call_tool("mark_task_completed", {
                        "task_id": task_id, "final_output": output
                    })

Running Tests

pip install -e ".[dev]"
pytest tests/ -v

All tests use an in-memory SQLite database and require no external services.