planningmcp

Planning and execution engine for AI agents with MCP support

Part of the MCP AI Suite.

Features

• Goal decomposition -- break goals into executable steps via pattern, template, LLM, or hybrid decomposers
• Plan execution with step-by-step tracking, parallel execution, and progress reporting
• Replanning -- automatic recovery from step failures via retry, skip, or LLM-based replan strategies
• Cost estimation and budget management with per-plan and per-namespace limits
• Plan optimization -- auto-parallelization, bottleneck detection, and critical path analysis
• Plan templates -- 8 built-in workflows: deploy, migrate, audit, analyze, setup, refactor, onboard, incident
• LTP integration -- compile goals into Lean Task Protocol plans for deterministic execution (provided by the separate ltpmcp dependency, whose LTP types planningmcp re-exports)
• Mermaid diagram export for plan visualization
• OpenTelemetry tracing support (optional)

Installation

pip install mcpaisuite-planningmcp
# Optional extras:
pip install mcpaisuite-planningmcp[dev]       # Development tools
pip install mcpaisuite-planningmcp[all]       # All integrations
pip install mcpaisuite-planningmcp[litellm]   # LLM-based decomposition
pip install mcpaisuite-planningmcp[redis]     # Redis plan store
pip install mcpaisuite-planningmcp[tracing]   # OpenTelemetry tracing

Quick Start

from planningmcp import PlanningFactory

planner = PlanningFactory.create(plan_store="sqlite", sqlite_path="plans.db")
plan = await planner.create_plan("Research and summarize Python 3.13 features")
plan = await planner.execute_plan(plan.id)
print(f"Status: {plan.status}, Progress: {plan.progress:.0%}")

MCP Server

planningmcp serve

Configuration

API Reference

PlanningPipeline

Central orchestrator for plan creation, execution, and learning.

await planner.create_plan(goal, namespace="default", context="") -> Plan
await planner.execute_plan(plan_id) -> Plan
await planner.execute_step(plan_id, step_id) -> StepResult
await planner.estimate_cost(plan_id) -> CostEstimate
await planner.replan(plan_id, failed_step_id) -> Plan
await planner.optimize_plan(plan_id) -> Plan
await planner.get_bottlenecks(plan_id) -> list[dict]
await planner.get_critical_path(plan_id) -> list[dict]
await planner.to_mermaid(plan_id) -> str
await planner.compile_ltp(goal) -> LTPPlan
await planner.run_ltp(goal, tool_executor=None) -> dict

PlanningFactory

PlanningFactory.default()                # SQLite store (planningmcp.db), template decomposer
PlanningFactory.from_env()               # Build from environment variables
PlanningFactory.from_yaml("config.yaml") # Build from YAML config
PlanningFactory.create(decomposer="hybrid", plan_store="sqlite", completion_fn=my_llm, ...)

Architecture

PlanningPipeline coordinates goal decomposition (via pluggable decomposers), step execution (via PlanExecutor with parallel support), and learning (via PlanHistory). ReplanEngine handles failure recovery with configurable strategies. BudgetManager enforces cost limits, and the optional memory/RAG integrations enrich plan context from prior executions. Plans are stored in pluggable backends (memory, SQLite, Redis).

Testing

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

License

Apache-2.0 — see LICENSE.

Open source for individuals and open-source projects. For commercial use in closed-source products, a commercial license is available — contact contact@mcpaisuite.com.