Skip to content

Latest commit

 

History

History
117 lines (93 loc) · 3.5 KB

File metadata and controls

117 lines (93 loc) · 3.5 KB

MCP Tasks Workflow Engine – Primitive for Long-Running Agentic Work - App Specification

Overview

This specification outlines the MCP Tasks Workflow Engine, a reusable workflow component (pip 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. This tool is designed for framework authors and agent builders who need to run durable jobs like code migrations, data exports, or LLM fine-tuning via MCP without reinventing orchestration.

Problem Statement: Developers currently build ad-hoc polling loops or custom queues for long-lived agent tasks, wasting tokens and complicating error handling.

Target Audience: Framework authors and agent builders (e.g., using Claude Agent SDK, Microsoft Agent Framework, or LangChain) who need to run durable jobs.

Tech Stack

  • Python 3.11+
  • click
  • pytest

Environment Setup

Prerequisites

  • Python 3.11+
  • pip

Configuration

Environment Variable Default Value Description
MCP_TASKS_DB tasks.db Path to the SQLite database

Architecture

+-----------------+
|  mcp_tasks.py   |
+-----------------+
       |   
+------+------+
|   Task   |  CLI  |
+------+------+
       |
+-----------------+
|  database.py    |
+-----------------+

Core Features

Feature 1: Task Creation

Description: Create a new task with a unique ID and initial state.

Requirements:

  • Accept a task name, description, and initial state.
  • Store the task in the SQLite database.

Test Steps:

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

Feature 2: Task Monitoring

Description: Retrieve the status of a task.

Requirements:

  • Accept a task ID.
  • Return the task's name, description, and current state.

Test Steps:

  1. mcp-tasks status 1 -> Task ID: 1, Name: Code Migration, Description: Migrate codebase to new framework, State: pending

Feature 3: Task Cancellation

Description: Cancel a running task.

Requirements:

  • Accept a task ID.
  • Update the task's state to "cancelled".

Test Steps:

  1. mcp-tasks cancel 1 -> Task ID: 1 cancelled

Data Models

from pydantic import BaseModel

class Task(BaseModel):
    id: int
    name: str
    description: str
    state: str

File Structure

mcp_tasks/
├── mcp_tasks.py
├── database.py
├── __init__.py
├── main.py
├── cli.py
├── README.md
├── setup.py
└── tests/
    ├── test_core.py
    └── test_cli.py

Test Plan

  • tests/test_core.py::test_create_task
  • tests/test_core.py::test_get_task_status
  • tests/test_core.py::test_cancel_task
  • tests/test_cli.py::test_create_task_cli
  • tests/test_cli.py::test_get_task_status_cli
  • tests/test_cli.py::test_cancel_task_cli

Success Criteria

  1. The mcp-tasks create command creates a new task in the SQLite database.
  2. The mcp-tasks status command retrieves the status of a task accurately.
  3. The mcp-tasks cancel command updates the task's state to "cancelled".
  4. All tests in tests/test_core.py and tests/test_cli.py pass.
  5. The tool can be installed and used entirely offline.

Constraints & Notes

  • No external API calls — all processing is local.
  • Target: working MVP in 5 build iterations.
  • Prioritize "works correctly" over "feature complete".