A reusable workflow component (pip package) that implements the MCP 'Tasks' primitive, providing APIs to create, monitor, and stream results from long-running agent jobs. It supports native streaming of incremental outputs, checkpointing, and cancellation, and can be plugged into any MCP host as a subprocess or external server.

Quick Start • Features • Examples • Contributing

What is this?

The MCP Tasks Workflow Engine is a reusable Python package that implements the MCP 'Tasks' primitive. It provides APIs to create, monitor, and stream results from long-running agent jobs, supporting native streaming of incremental outputs, checkpointing, and cancellation. It is designed for framework authors and agent builders who need to run durable jobs without reinventing orchestration.

$ mcp-tasks create "Code Migration" "Migrate codebase to new framework" "pending"
Task created with ID: 1

Problem

While the MCP roadmap defines a Tasks primitive for asynchronous work, developers still build ad-hoc polling loops or custom queues for long-lived agent tasks, wasting tokens and complicating error handling. No off-the-shelf component exists to bring this primitive to agents.

Features

Feature	Description
Task Creation	Create a new task with a unique ID and initial state.
Task Monitoring	Retrieve the status of a task by its ID.
Task Cancellation	Cancel a running task and update its state.
Native Streaming	Stream incremental outputs from long-running tasks.
Checkpointing	Save and restore task state to manage long-running processes.
Offline Operation	All processing is local, with no external API calls.
Extensible	Can be plugged into any MCP host as a subprocess or external server.

Quick Start

1. Clone the repository:

   $ git clone https://github.com/your-repo/mcp-tasks-workflow-engine.git
   $ cd mcp-tasks-workflow-engine

2. Install the package:

   $ pip install .

3. Create a new task:

   $ mcp-tasks create "Code Migration" "Migrate codebase to new framework" "pending"
   Task created with ID: 1

Examples

Create a new task

Command:

$ mcp-tasks create "Data Export" "Export user data to CSV" "in_progress"

Output:

Task created with ID: 2

Retrieve the status of a task

Command:

$ mcp-tasks status 2

Output:

Task ID: 2, Name: Data Export, Description: Export user data to CSV, State: in_progress

Cancel a task

Command:

$ mcp-tasks cancel 2

Output:

Task ID: 2 cancelled

File Structure

MCP Tasks Workflow Engine – Primitive for Long-Running Agentic Work/
  mcp_tasks/
    __init__.py          # Package initializer
    __main__.py          # Main entry point
    cli.py               # Command-line interface
    database.py          # Database operations
    main.py              # Core logic
    mcp_tasks.py         # Main class definitions
  tests/
    __init__.py          # Test initialization
    test_cli.py          # CLI test suite
    test_core.py         # Core logic test suite
  .gitignore             # Git ignore rules
  .heartbeat-callback    # Heartbeat callback config
  README.md              # Project documentation
  conftest.py            # Pytest configuration
  pytest.ini             # Pytest configuration file
  requirements.txt       # Python dependencies
  setup.py               # Package setup
  spec.md                # Detailed specification

Tech Stack

Technology	Purpose
Python 3.11+	Core language
click	CLI framework
pytest	Testing framework

Contributing

1. Fork the repository.
2. Make your changes and add tests.
3. Ensure all tests pass.
4. Submit a pull request.

License

MIT

Author

Matthew Snow -- M2AI | @m2ai-portfolio