Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
63 changes: 45 additions & 18 deletions python/src/mcp_server/features/tasks/task_tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@

import json
import logging
from typing import Any, Dict, List, Optional, TypedDict
from typing import Any, Dict, List, Optional
from urllib.parse import urljoin

import httpx
Expand All @@ -20,19 +20,6 @@
logger = logging.getLogger(__name__)


class TaskUpdateFields(TypedDict, total=False):
"""Valid fields that can be updated on a task."""

title: str
description: str
status: str # "todo" | "doing" | "review" | "done"
assignee: str # "User" | "Archon" | "AI IDE Agent" | "prp-executor" | "prp-validator"
task_order: int # 0-100, higher = more priority
feature: Optional[str]
sources: Optional[List[Dict[str, str]]]
code_examples: Optional[List[Dict[str, str]]]


def register_task_tools(mcp: FastMCP):
"""Register individual task management tools with the MCP server."""

Expand Down Expand Up @@ -315,26 +302,66 @@ async def get_task(ctx: Context, task_id: str) -> str:
async def update_task(
ctx: Context,
task_id: str,
update_fields: TaskUpdateFields,
title: Optional[str] = None,
description: Optional[str] = None,
status: Optional[str] = None,
assignee: Optional[str] = None,
task_order: Optional[int] = None,
feature: Optional[str] = None,
sources: Optional[List[Dict[str, str]]] = None,
code_examples: Optional[List[Dict[str, str]]] = None,
) -> str:
"""
Update a task's properties.

Args:
task_id: UUID of the task to update
update_fields: Dict of fields to update (e.g., {"status": "doing", "assignee": "AI IDE Agent"})
title: New title (optional)
description: New description (optional)
status: New status - "todo" | "doing" | "review" | "done" (optional)
assignee: New assignee (optional)
task_order: New priority order (optional)
feature: New feature label (optional)
sources: New source references (optional)
code_examples: New code examples (optional)

Returns:
JSON with updated task details

Examples:
update_task(task_id="uuid", update_fields={"status": "doing"})
update_task(task_id="uuid", update_fields={"title": "New Title", "description": "Updated description"})
update_task(task_id="uuid", status="doing")
update_task(task_id="uuid", title="New Title", description="Updated description")
"""
try:
api_url = get_api_url()
timeout = get_default_timeout()

# Build update_fields dict from provided parameters
update_fields = {}
if title is not None:
update_fields["title"] = title
if description is not None:
update_fields["description"] = description
if status is not None:
update_fields["status"] = status
if assignee is not None:
update_fields["assignee"] = assignee
if task_order is not None:
update_fields["task_order"] = task_order
if feature is not None:
update_fields["feature"] = feature
if sources is not None:
update_fields["sources"] = sources
if code_examples is not None:
update_fields["code_examples"] = code_examples

if not update_fields:
return MCPErrorFormatter.format_error(
error_type="validation_error",
message="No fields to update",
suggestion="Provide at least one field to update",
)

async with httpx.AsyncClient(timeout=timeout) as client:
response = await client.put(
urljoin(api_url, f"/api/tasks/{task_id}"), json=update_fields
Expand Down
99 changes: 58 additions & 41 deletions python/src/mcp_server/mcp_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -192,57 +192,74 @@ async def lifespan(server: FastMCP) -> AsyncIterator[ArchonContext]:
# Archon MCP Server Instructions

## 🚨 CRITICAL RULES (ALWAYS FOLLOW)
1. **Task Management**: ALWAYS use Archon MCP tools for task management,
You can combine them with your TODO tools but always make sure that the first todo is to update archon
and the last todo is to update archon.

Example: Use TodoWrite to create a set of new todos
[]Create the task in archon
[]Research deeply using archon rag
[]Research on the web using websearch tools
[]Deeply look into the codebase patterns and integration points
[]Update Archon tasks with the findings
[]Create implementation tasks in Archon

This is to ensure efficient task management and collaboration.
Making sure all critical details are in Archon.

You can think of it as Archon is where you manage the task that needs to be shared with the team
And your todo is your internal subtasks/todos that does not need to be shared with the team.
1. **Task Management**: ALWAYS use Archon MCP tools for task management.
- Combine with your local TODO tools for granular tracking
- First TODO: Update Archon task status
- Last TODO: Update Archon with findings/completion

2. **Research First**: Before implementing, use perform_rag_query and search_code_examples
3. **Task-Driven Development**: Never code without checking current tasks first

## 📋 Core Workflow
For every coding task, follow this cycle:
1. Check current task: manage_task(action="get", task_id="...")
2. Research: perform_rag_query() + search_code_examples()
3. Update to doing: manage_task(action="update", update_fields={"status": "doing"})
4. Implement based on research findings
5. Mark for review: manage_task(action="update", update_fields={"status": "review"})
6. Get next task: manage_task(action="list", filter_by="status", filter_value="todo")

## 🏗️ Project Initialization
- New project: manage_project(action="create", title="...", prd={...})
- Existing project: manage_task(action="list", filter_by="project", filter_value="...")
- Always create atomic tasks (1-4 hours of work each)

### Task Management Cycle
1. **Get current task**: `get_task(task_id="...")`
2. **Mark as doing**: `update_task(task_id="...", status="doing")`
3. **Research phase**:
- `perform_rag_query(query="...", match_count=5)`
- `search_code_examples(query="...", match_count=3)`
4. **Implementation**: Code based on research findings
5. **Mark for review**: `update_task(task_id="...", status="review")`
6. **Get next task**: `list_tasks(filter_by="status", filter_value="todo")`

### Available Task Functions
- `create_task(project_id, title, description, assignee="User", ...)`
- `list_tasks(filter_by="status", filter_value="todo", project_id=None)`
- `get_task(task_id)`
- `update_task(task_id, title=None, status=None, assignee=None, ...)`
- `delete_task(task_id)`

## 🏗️ Project Management

### Project Functions
- `create_project(title, description, github_repo=None)`
- `list_projects()`
- `get_project(project_id)`
- `update_project(project_id, title=None, description=None, ...)`
- `delete_project(project_id)`

### Document Functions
- `create_document(project_id, title, document_type, content=None, ...)`
- `list_documents(project_id)`
- `get_document(project_id, doc_id)`
- `update_document(project_id, doc_id, title=None, content=None, ...)`
- `delete_document(project_id, doc_id)`

## 🔍 Research Patterns
- Architecture: perform_rag_query(query="[tech] patterns", match_count=5)
- Implementation: search_code_examples(query="[feature] example", match_count=3)
- Keep match_count around (5) for focused results
- Combine RAG with websearch tools for better results
- **Architecture patterns**: `perform_rag_query(query="[tech] architecture patterns", match_count=5)`
- **Code examples**: `search_code_examples(query="[feature] implementation", match_count=3)`
- **Source discovery**: `get_available_sources()`
- Keep match_count around 3-5 for focused results

## 📊 Task Status Flow
todo → doing → review → done
- Only one task in 'doing' at a time
`todo``doing``review``done`
- Only ONE task in 'doing' status at a time
- Use 'review' for completed work awaiting validation
- Archive obsolete tasks

## 💾 Version Control
- All documents auto-versioned on update
- Use manage_versions to view history or restore
- Deletions preserve version history
- Mark tasks 'done' only after verification

## 💾 Version Management
- `create_version(project_id, field_name, content, change_summary)`
- `list_versions(project_id, field_name=None)`
- `get_version(project_id, field_name, version_number)`
- `restore_version(project_id, field_name, version_number)`
- Field names: "docs", "features", "data", "prd"

## 🎯 Best Practices
1. **Atomic Tasks**: Create tasks that take 1-4 hours
2. **Clear Descriptions**: Include acceptance criteria in task descriptions
3. **Use Features**: Group related tasks with feature labels
4. **Add Sources**: Link relevant documentation to tasks
5. **Track Progress**: Update task status as you work
"""

# Initialize the main FastMCP server with fixed configuration
Expand Down
29 changes: 28 additions & 1 deletion python/tests/mcp_server/features/tasks/test_task_tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -168,12 +168,39 @@ async def test_update_task_status(mock_mcp, mock_context):
mock_client.return_value.__aenter__.return_value = mock_async_client

result = await update_task(
mock_context, task_id="task-123", update_fields={"status": "doing", "assignee": "User"}
mock_context, task_id="task-123", status="doing", assignee="User"
)

result_data = json.loads(result)
assert result_data["success"] is True
assert "Task updated successfully" in result_data["message"]

# Verify the PUT request was made with correct data
call_args = mock_async_client.put.call_args
sent_data = call_args[1]["json"]
assert sent_data["status"] == "doing"
assert sent_data["assignee"] == "User"


@pytest.mark.asyncio
async def test_update_task_no_fields(mock_mcp, mock_context):
"""Test updating task with no fields returns validation error."""
register_task_tools(mock_mcp)

# Get the update_task function
update_task = mock_mcp._tools.get("update_task")

assert update_task is not None, "update_task tool not registered"

# Call update_task with no optional fields
result = await update_task(mock_context, task_id="task-123")

result_data = json.loads(result)
assert result_data["success"] is False
assert "error" in result_data
assert isinstance(result_data["error"], dict)
assert result_data["error"]["type"] == "validation_error"
assert "No fields to update" in result_data["error"]["message"]


@pytest.mark.asyncio
Expand Down