luis.erlacher/Archon
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: MCP server optimization with tool consolidation and vertical sl… (#647)

Browse Source 
* feat: MCP server optimization with tool consolidation and vertical slice architecture

- Consolidated MCP tools from ~20 to 8 tools for improved UX
- Restructured to vertical slice architecture (features/domain pattern)
- Optimized payload sizes with truncation and array count replacements
- Changed default include_closed to true for better task visibility
- Moved RAG module to features directory structure
- Removed legacy modules directory in favor of feature-based organization

Key improvements:
- list_tasks, manage_task (create/update/delete consolidated)
- list_projects, manage_project (create/update/delete consolidated)
- list_documents, manage_document (create/update/delete consolidated)
- list_versions, manage_version (create/restore consolidated)
- Reduced default page size from 50 to 10 items
- Added search query support to list operations

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>

* feat: Consolidate MCP tools and rename list_* to find_*

Major refactoring of MCP tools to reduce complexity and improve naming:

## Tool Consolidation (22 → ~10 tools)
- Consolidated CRUD operations into two tools per domain:
  - find_[resource]: Handles list, search, and get single item
  - manage_[resource]: Handles create, update, delete with "action" parameter
- Removed backward compatibility/legacy function mappings
- Optimized response payloads with truncation (1000 char limit for projects/tasks)

## Renamed Functions
- list_projects → find_projects
- list_tasks → find_tasks
- list_documents → find_documents
- list_versions → find_versions

## Bug Fixes
- Fixed supabase query chaining bug where .or_() calls overwrote previous conditions
- Fixed search implementation to handle single vs multiple terms correctly

## Test Updates
- Updated all tests to use new consolidated tools
- Removed problematic test_consolidated_tools.py
- Fixed error type assertions to match actual responses
- All 44 tests passing

## Documentation Updates
- Updated CLAUDE.md with new tool names and patterns
- Updated MCP instructions with consolidated tool examples
- Added guidance to avoid backward compatibility code

## API Changes
- Updated API route defaults: include_closed=True, per_page=10
- Aligned defaults with consolidated tool implementations

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>

---------

Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
Cole Medin 2025-09-13 10:52:14 -05:00 committed by GitHub
parent ce2f871ebb
commit 34a51ec362
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
16 changed files with 1115 additions and 1320 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
CLAUDE.md
View File

@ -98,6 +98,7 @@ def process_batch(items):
### Code Quality
- Remove dead code immediately rather than maintaining it - no backward compatibility or legacy functions
- Avoid backward compatibility mappings or legacy function wrappers
- Prioritize functionality over production-ready patterns
- Focus on user experience and feature completeness
- When updating code, don't reference what is changing (avoid keywords like LEGACY, CHANGED, REMOVED), instead focus on comments that document just the functionality of the code
@ -383,6 +384,15 @@ ARCHON_UI_PORT=3737 # Frontend port
4. Use TanStack Query hook from `src/features/[feature]/hooks/`
5. Apply Tron-inspired glassmorphism styling with Tailwind
### Add or modify MCP tools
1. MCP tools are in `python/src/mcp_server/features/[feature]/[feature]_tools.py`
2. Follow the pattern:
- `find_[resource]` - Handles list, search, and get single item operations
- `manage_[resource]` - Handles create, update, delete with an "action" parameter
3. Optimize responses by truncating/filtering fields in list operations
4. Register tools in the feature's `__init__.py` file
### Debug MCP connection issues
1. Check MCP health: `curl http://localhost:8051/health`
@ -421,16 +431,28 @@ npm run lint:files src/components/SomeComponent.tsx
## MCP Tools Available
When connected to Client/Cursor/Windsurf:
When connected to Claude/Cursor/Windsurf, the following tools are available:
- `archon:perform_rag_query` - Search knowledge base
- `archon:search_code_examples` - Find code snippets
- `archon:create_project` - Create new project
- `archon:list_projects` - List all projects
- `archon:create_task` - Create task in project
- `archon:list_tasks` - List and filter tasks
- `archon:update_task` - Update task status/details
- `archon:get_available_sources` - List knowledge sources
### Knowledge Base Tools
- `archon:rag_search_knowledge_base` - Search knowledge base for relevant content
- `archon:rag_search_code_examples` - Find code snippets in the knowledge base
- `archon:rag_get_available_sources` - List available knowledge sources
### Project Management
- `archon:find_projects` - Find all projects, search, or get specific project (by project_id)
- `archon:manage_project` - Manage projects with actions: "create", "update", "delete"
### Task Management
- `archon:find_tasks` - Find tasks with search, filters, or get specific task (by task_id)
- `archon:manage_task` - Manage tasks with actions: "create", "update", "delete"
### Document Management
- `archon:find_documents` - Find documents, search, or get specific document (by document_id)
- `archon:manage_document` - Manage documents with actions: "create", "update", "delete"
### Version Control
- `archon:find_versions` - Find version history or get specific version
- `archon:manage_version` - Manage versions with actions: "create", "restore"
## Important Notes

473
python/src/mcp_server/features/documents/document_tools.py
View File

@ -1,8 +1,7 @@
"""
Simple document management tools for Archon MCP Server.
Consolidated document management tools for Archon MCP Server.
Provides separate, focused tools for each document operation.
Supports various document types including specs, designs, notes, and PRPs.
Reduces the number of individual CRUD operations while maintaining full functionality.
"""
import json
@ -19,309 +18,265 @@ from src.server.config.service_discovery import get_api_url
logger = logging.getLogger(__name__)
# Optimization constants
DEFAULT_PAGE_SIZE = 10
def optimize_document_response(doc: dict) -> dict:
"""Optimize document object for MCP response."""
doc = doc.copy() # Don't modify original
# Remove full content in list views
if "content" in doc:
del doc["content"]
return doc
def register_document_tools(mcp: FastMCP):
"""Register individual document management tools with the MCP server."""
"""Register consolidated document management tools with the MCP server."""
@mcp.tool()
async def create_document(
async def find_documents(
ctx: Context,
project_id: str,
title: str,
document_type: str,
content: dict[str, Any] | None = None,
tags: list[str] | None = None,
author: str | None = None,
document_id: str | None = None, # For getting single document
query: str | None = None, # Search capability
document_type: str | None = None, # Filter by type
page: int = 1,
per_page: int = DEFAULT_PAGE_SIZE,
) -> str:
"""
Create a new document with automatic versioning.
Find and search documents (consolidated: list + search + get).
Args:
project_id: Project UUID (required)
title: Document title (required)
document_type: Type of document. Common types:
- "spec": Technical specifications
- "design": Design documents
- "note": General notes
- "prp": Product requirement prompts
- "api": API documentation
- "guide": User guides
content: Document content as structured JSON (optional).
Can be any JSON structure that fits your needs.
tags: List of tags for categorization (e.g., ["backend", "auth"])
author: Document author name (optional)
document_id: Get specific document (returns full content)
query: Search in title/content
document_type: Filter by type (spec/design/note/prp/api/guide)
page: Page number for pagination
per_page: Items per page (default: 10)
Returns:
JSON with document details:
{
"success": true,
"document": {...},
"document_id": "doc-123",
"message": "Document created successfully"
}
JSON array of documents or single document
Examples:
# Create API specification
create_document(
project_id="550e8400-e29b-41d4-a716-446655440000",
title="REST API Specification",
document_type="spec",
content={
"endpoints": [
{"path": "/users", "method": "GET", "description": "List users"},
{"path": "/users/{id}", "method": "GET", "description": "Get user"}
],
"authentication": "Bearer token",
"version": "1.0.0"
},
tags=["api", "backend"],
author="API Team"
)
# Create design document
create_document(
project_id="550e8400-e29b-41d4-a716-446655440000",
title="Authentication Flow Design",
document_type="design",
content={
"overview": "OAuth2 implementation design",
"components": ["AuthProvider", "TokenManager", "UserSession"],
"flow": {"step1": "Redirect to provider", "step2": "Exchange code"}
}
)
find_documents(project_id="p-1") # All project docs
find_documents(project_id="p-1", query="api") # Search
find_documents(project_id="p-1", document_id="d-1") # Get one
find_documents(project_id="p-1", document_type="spec") # Filter
"""
try:
api_url = get_api_url()
timeout = get_default_timeout()
# Single document get mode
if document_id:
async with httpx.AsyncClient(timeout=timeout) as client:
response = await client.get(
urljoin(api_url, f"/api/projects/{project_id}/docs/{document_id}")
)
if response.status_code == 200:
document = response.json()
# Don't optimize single document - return full content
return json.dumps({"success": True, "document": document})
elif response.status_code == 404:
return MCPErrorFormatter.format_error(
error_type="not_found",
message=f"Document {document_id} not found",
suggestion="Verify the document ID is correct",
http_status=404,
)
else:
return MCPErrorFormatter.from_http_error(response, "get document")
# List mode
async with httpx.AsyncClient(timeout=timeout) as client:
response = await client.post(
urljoin(api_url, f"/api/projects/{project_id}/docs"),
json={
"document_type": document_type,
"title": title,
"content": content or {},
"tags": tags,
"author": author,
},
)
if response.status_code == 200:
result = response.json()
return json.dumps({
"success": True,
"document": result.get("document"),
"document_id": result.get("document", {}).get("id"),
"message": result.get("message", "Document created successfully"),
})
else:
return MCPErrorFormatter.from_http_error(response, "create document")
except httpx.RequestError as e:
return MCPErrorFormatter.from_exception(
e, "create document", {"project_id": project_id, "title": title}
)
except Exception as e:
logger.error(f"Error creating document: {e}", exc_info=True)
return MCPErrorFormatter.from_exception(e, "create document")
@mcp.tool()
async def list_documents(ctx: Context, project_id: str) -> str:
"""
List all documents for a project.
Args:
project_id: Project UUID (required)
Returns:
JSON array of documents
Example:
list_documents(project_id="uuid")
"""
try:
api_url = get_api_url()
timeout = get_default_timeout()
async with httpx.AsyncClient(timeout=timeout) as client:
# Pass include_content=False for lightweight response
response = await client.get(
urljoin(api_url, f"/api/projects/{project_id}/docs"),
params={"include_content": False}
urljoin(api_url, f"/api/projects/{project_id}/docs")
)
if response.status_code == 200:
result = response.json()
data = response.json()
documents = data.get("documents", [])
# Apply filters
if document_type:
documents = [d for d in documents if d.get("document_type") == document_type]
if query:
query_lower = query.lower()
documents = [
d for d in documents
if query_lower in d.get("title", "").lower()
or query_lower in str(d.get("content", "")).lower()
]
# Apply pagination
start_idx = (page - 1) * per_page
end_idx = start_idx + per_page
paginated = documents[start_idx:end_idx]
# Optimize document responses - remove content from list views
optimized = [optimize_document_response(d) for d in paginated]
return json.dumps({
"success": True,
"documents": result.get("documents", []),
"count": len(result.get("documents", [])),
"documents": optimized,
"count": len(optimized),
"total": len(documents),
"project_id": project_id,
"query": query,
"document_type": document_type
})
else:
return MCPErrorFormatter.from_http_error(response, "list documents")
except httpx.RequestError as e:
return MCPErrorFormatter.from_exception(e, "list documents", {"project_id": project_id})
return MCPErrorFormatter.from_exception(e, "list documents")
except Exception as e:
logger.error(f"Error listing documents: {e}", exc_info=True)
return MCPErrorFormatter.from_exception(e, "list documents")
@mcp.tool()
async def get_document(ctx: Context, project_id: str, doc_id: str) -> str:
"""
Get detailed information about a specific document.
Args:
project_id: Project UUID (required)
doc_id: Document UUID (required)
Returns:
JSON with complete document details
Example:
get_document(project_id="uuid", doc_id="doc-uuid")
"""
try:
api_url = get_api_url()
timeout = get_default_timeout()
async with httpx.AsyncClient(timeout=timeout) as client:
response = await client.get(
urljoin(api_url, f"/api/projects/{project_id}/docs/{doc_id}")
)
if response.status_code == 200:
document = response.json()
return json.dumps({"success": True, "document": document})
elif response.status_code == 404:
return MCPErrorFormatter.format_error(
error_type="not_found",
message=f"Document {doc_id} not found",
suggestion="Verify the document ID is correct and exists in this project",
http_status=404,
)
else:
return MCPErrorFormatter.from_http_error(response, "get document")
except httpx.RequestError as e:
return MCPErrorFormatter.from_exception(
e, "get document", {"project_id": project_id, "doc_id": doc_id}
)
except Exception as e:
logger.error(f"Error getting document: {e}", exc_info=True)
return MCPErrorFormatter.from_exception(e, "get document")
@mcp.tool()
async def update_document(
async def manage_document(
ctx: Context,
action: str, # "create" | "update" | "delete"
project_id: str,
doc_id: str,
document_id: str | None = None,
title: str | None = None,
document_type: str | None = None,
content: dict[str, Any] | None = None,
tags: list[str] | None = None,
author: str | None = None,
) -> str:
"""
Update a document's properties.
Manage documents (consolidated: create/update/delete).
Args:
action: "create" | "update" | "delete"
project_id: Project UUID (required)
doc_id: Document UUID (required)
title: New document title (optional)
content: New document content (optional)
tags: New tags list (optional)
author: New author (optional)
Returns:
JSON with updated document details
Example:
update_document(project_id="uuid", doc_id="doc-uuid", title="New Title",
content={"updated": "content"})
document_id: Document UUID for update/delete
title: Document title
document_type: spec/design/note/prp/api/guide
content: Structured JSON content
tags: List of tags (e.g. ["backend", "auth"])
author: Document author name
Examples:
manage_document("create", project_id="p-1", title="API Spec", document_type="spec")
manage_document("update", project_id="p-1", document_id="d-1", content={...})
manage_document("delete", project_id="p-1", document_id="d-1")
Returns: {success: bool, document?: object, message: string}
"""
try:
api_url = get_api_url()
timeout = get_default_timeout()
# Build update fields
update_fields: dict[str, Any] = {}
if title is not None:
update_fields["title"] = title
if content is not None:
update_fields["content"] = content
if tags is not None:
update_fields["tags"] = tags
if author is not None:
update_fields["author"] = author
async with httpx.AsyncClient(timeout=timeout) as client:
response = await client.put(
urljoin(api_url, f"/api/projects/{project_id}/docs/{doc_id}"),
json=update_fields,
)
if response.status_code == 200:
result = response.json()
return json.dumps({
"success": True,
"document": result.get("document"),
"message": result.get("message", "Document updated successfully"),
})
else:
return MCPErrorFormatter.from_http_error(response, "update document")
except httpx.RequestError as e:
return MCPErrorFormatter.from_exception(
e, "update document", {"project_id": project_id, "doc_id": doc_id}
)
except Exception as e:
logger.error(f"Error updating document: {e}", exc_info=True)
return MCPErrorFormatter.from_exception(e, "update document")
@mcp.tool()
async def delete_document(ctx: Context, project_id: str, doc_id: str) -> str:
"""
Delete a document.
Args:
project_id: Project UUID (required)
doc_id: Document UUID (required)
Returns:
JSON confirmation of deletion
Example:
delete_document(project_id="uuid", doc_id="doc-uuid")
"""
try:
api_url = get_api_url()
timeout = get_default_timeout()
async with httpx.AsyncClient(timeout=timeout) as client:
response = await client.delete(
urljoin(api_url, f"/api/projects/{project_id}/docs/{doc_id}")
)
if response.status_code == 200:
result = response.json()
return json.dumps({
"success": True,
"message": result.get("message", f"Document {doc_id} deleted successfully"),
})
elif response.status_code == 404:
return MCPErrorFormatter.format_error(
error_type="not_found",
message=f"Document {doc_id} not found",
suggestion="Verify the document ID is correct and exists in this project",
http_status=404,
if action == "create":
if not title or not document_type:
return MCPErrorFormatter.format_error(
"validation_error",
"title and document_type required for create"
)
response = await client.post(
urljoin(api_url, f"/api/projects/{project_id}/docs"),
json={
"title": title,
"document_type": document_type,
"content": content or {},
"tags": tags or [],
"author": author or "User",
}
)
if response.status_code == 200:
result = response.json()
document = result.get("document")
# Don't optimize for create - return full document
return json.dumps({
"success": True,
"document": document,
"document_id": document.get("id") if document else None,
"message": result.get("message", "Document created successfully")
})
else:
return MCPErrorFormatter.from_http_error(response, "create document")
elif action == "update":
if not document_id:
return MCPErrorFormatter.format_error(
"validation_error",
"document_id required for update"
)
update_data = {}
if title is not None:
update_data["title"] = title
if content is not None:
update_data["content"] = content
if tags is not None:
update_data["tags"] = tags
if author is not None:
update_data["author"] = author
if not update_data:
return MCPErrorFormatter.format_error(
"validation_error",
"No fields to update"
)
response = await client.put(
urljoin(api_url, f"/api/projects/{project_id}/docs/{document_id}"),
json=update_data
)
if response.status_code == 200:
result = response.json()
document = result.get("document")
# Don't optimize for update - return full document
return json.dumps({
"success": True,
"document": document,
"message": result.get("message", "Document updated successfully")
})
else:
return MCPErrorFormatter.from_http_error(response, "update document")
elif action == "delete":
if not document_id:
return MCPErrorFormatter.format_error(
"validation_error",
"document_id required for delete"
)
response = await client.delete(
urljoin(api_url, f"/api/projects/{project_id}/docs/{document_id}")
)
if response.status_code == 200:
result = response.json()
return json.dumps({
"success": True,
"message": result.get("message", "Document deleted successfully")
})
else:
return MCPErrorFormatter.from_http_error(response, "delete document")
else:
return MCPErrorFormatter.from_http_error(response, "delete document")
return MCPErrorFormatter.format_error(
"invalid_action",
f"Unknown action: {action}"
)
except httpx.RequestError as e:
return MCPErrorFormatter.from_exception(
e, "delete document", {"project_id": project_id, "doc_id": doc_id}
)
return MCPErrorFormatter.from_exception(e, f"{action} document")
except Exception as e:
logger.error(f"Error deleting document: {e}", exc_info=True)
return MCPErrorFormatter.from_exception(e, "delete document")
logger.error(f"Error managing document ({action}): {e}", exc_info=True)
return MCPErrorFormatter.from_exception(e, f"{action} document")

445
python/src/mcp_server/features/documents/version_tools.py
View File

@ -1,8 +1,7 @@
"""
Simple version management tools for Archon MCP Server.
Consolidated version management tools for Archon MCP Server.
Provides separate, focused tools for version control operations.
Supports versioning of documents, features, and other project data.
Reduces the number of individual CRUD operations while maintaining full functionality.
"""
import json
@ -19,328 +18,216 @@ from src.server.config.service_discovery import get_api_url
logger = logging.getLogger(__name__)
# Optimization constants
DEFAULT_PAGE_SIZE = 10
def optimize_version_response(version: dict) -> dict:
"""Optimize version object for MCP response."""
version = version.copy() # Don't modify original
# Remove content in list views - it's too large
if "content" in version:
del version["content"]
return version
def register_version_tools(mcp: FastMCP):
"""Register individual version management tools with the MCP server."""
"""Register consolidated version management tools with the MCP server."""
@mcp.tool()
async def create_version(
async def find_versions(
ctx: Context,
project_id: str,
field_name: str,
content: dict[str, Any] | list[dict[str, Any]],
change_summary: str | None = None,
document_id: str | None = None,
created_by: str = "system",
field_name: str | None = None,
version_number: int | None = None, # For getting specific version
page: int = 1,
per_page: int = DEFAULT_PAGE_SIZE,
) -> str:
"""
Create a new version snapshot of project data.
Creates an immutable snapshot that can be restored later. The content format
depends on which field_name you're versioning.
Args:
project_id: Project UUID (e.g., "550e8400-e29b-41d4-a716-446655440000")
field_name: Which field to version - must be one of:
- "docs": For document arrays
- "features": For feature status objects
- "data": For general data objects
- "prd": For product requirement documents
content: Complete content to snapshot. Format depends on field_name:
For "docs" - pass array of document objects:
[{"id": "doc-123", "title": "API Guide", "content": {...}}]
For "features" - pass dictionary of features:
{"auth": {"status": "done"}, "api": {"status": "in_progress"}}
For "data" - pass any JSON object:
{"config": {"theme": "dark"}, "settings": {...}}
For "prd" - pass PRD object:
{"vision": "...", "features": [...], "metrics": [...]}
change_summary: Description of what changed (e.g., "Added OAuth docs")
document_id: Optional - for versioning specific doc in docs array
created_by: Who created this version (default: "system")
Returns:
JSON with version details:
{
"success": true,
"version": {"version_number": 3, "field_name": "docs"},
"message": "Version created successfully"
}
Examples:
# Version documents
create_version(
project_id="550e8400-e29b-41d4-a716-446655440000",
field_name="docs",
content=[{"id": "doc-1", "title": "Guide", "content": {"text": "..."}}],
change_summary="Updated user guide"
)
# Version features
create_version(
project_id="550e8400-e29b-41d4-a716-446655440000",
field_name="features",
content={"auth": {"status": "done"}, "api": {"status": "todo"}},
change_summary="Completed authentication"
)
"""
try:
api_url = get_api_url()
timeout = get_default_timeout()
async with httpx.AsyncClient(timeout=timeout) as client:
response = await client.post(
urljoin(api_url, f"/api/projects/{project_id}/versions"),
json={
"field_name": field_name,
"content": content,
"change_summary": change_summary,
"change_type": "manual",
"document_id": document_id,
"created_by": created_by,
},
)
if response.status_code == 200:
result = response.json()
version_num = result.get("version", {}).get("version_number")
return json.dumps({
"success": True,
"version": result.get("version"),
"version_number": version_num,
"message": f"Version {version_num} created successfully for {field_name} field",
})
elif response.status_code == 400:
error_text = response.text.lower()
if "invalid field_name" in error_text:
return MCPErrorFormatter.format_error(
error_type="validation_error",
message=f"Invalid field_name '{field_name}'. Must be one of: docs, features, data, or prd",
suggestion="Use one of the valid field names: docs, features, data, or prd",
http_status=400,
)
elif "content" in error_text and "required" in error_text:
return MCPErrorFormatter.format_error(
error_type="validation_error",
message="Content is required and cannot be empty. Provide the complete data to version.",
suggestion="Provide the complete data to version",
http_status=400,
)
elif "format" in error_text or "type" in error_text:
if field_name == "docs":
return MCPErrorFormatter.format_error(
error_type="validation_error",
message="For field_name='docs', content must be an array. Example: [{'id': 'doc1', 'title': 'Guide', 'content': {...}}]",
suggestion="Ensure content is an array of document objects",
http_status=400,
)
else:
return MCPErrorFormatter.format_error(
error_type="validation_error",
message=f"For field_name='{field_name}', content must be a dictionary/object. Example: {{'key': 'value'}}",
suggestion="Ensure content is a dictionary/object",
http_status=400,
)
return MCPErrorFormatter.format_error(
error_type="validation_error",
message=f"Invalid request: {response.text}",
suggestion="Check that all required fields are provided and valid",
http_status=400,
)
elif response.status_code == 404:
return MCPErrorFormatter.format_error(
error_type="not_found",
message=f"Project {project_id} not found",
suggestion="Please check the project ID is correct",
http_status=404,
)
else:
return MCPErrorFormatter.from_http_error(response, "create version")
except httpx.RequestError as e:
return MCPErrorFormatter.from_exception(
e, "create version", {"project_id": project_id, "field_name": field_name}
)
except Exception as e:
logger.error(f"Error creating version: {e}", exc_info=True)
return MCPErrorFormatter.from_exception(e, "create version")
@mcp.tool()
async def list_versions(ctx: Context, project_id: str, field_name: str | None = None) -> str:
"""
List version history for a project.
Find version history (consolidated: list + get).
Args:
project_id: Project UUID (required)
field_name: Filter by field name - "docs", "features", "data", "prd" (optional)
field_name: Filter by field (docs/features/data/prd)
version_number: Get specific version (requires field_name)
page: Page number for pagination
per_page: Items per page (default: 10)
Returns:
JSON array of versions with metadata
Example:
list_versions(project_id="uuid", field_name="docs")
JSON array of versions or single version
Examples:
find_versions(project_id="p-1") # All versions
find_versions(project_id="p-1", field_name="docs") # Doc versions
find_versions(project_id="p-1", field_name="docs", version_number=3) # Get v3
"""
try:
api_url = get_api_url()
timeout = get_default_timeout()
# Single version get mode
if field_name and version_number is not None:
async with httpx.AsyncClient(timeout=timeout) as client:
response = await client.get(
urljoin(api_url, f"/api/projects/{project_id}/versions/{field_name}/{version_number}")
)
if response.status_code == 200:
version = response.json()
# Don't optimize single version - return full details
return json.dumps({"success": True, "version": version})
elif response.status_code == 404:
return MCPErrorFormatter.format_error(
error_type="not_found",
message=f"Version {version_number} not found for field {field_name}",
suggestion="Verify the version number and field name",
http_status=404,
)
else:
return MCPErrorFormatter.from_http_error(response, "get version")
# List mode
params = {}
if field_name:
params["field_name"] = field_name
async with httpx.AsyncClient(timeout=timeout) as client:
response = await client.get(
urljoin(api_url, f"/api/projects/{project_id}/versions"), params=params
urljoin(api_url, f"/api/projects/{project_id}/versions"),
params=params
)
if response.status_code == 200:
result = response.json()
data = response.json()
versions = data.get("versions", [])
# Apply pagination
start_idx = (page - 1) * per_page
end_idx = start_idx + per_page
paginated = versions[start_idx:end_idx]
# Optimize version responses
optimized = [optimize_version_response(v) for v in paginated]
return json.dumps({
"success": True,
"versions": result.get("versions", []),
"count": len(result.get("versions", [])),
"versions": optimized,
"count": len(optimized),
"total": len(versions),
"project_id": project_id,
"field_name": field_name
})
else:
return MCPErrorFormatter.from_http_error(response, "list versions")
except httpx.RequestError as e:
return MCPErrorFormatter.from_exception(
e, "list versions", {"project_id": project_id, "field_name": field_name}
)
return MCPErrorFormatter.from_exception(e, "list versions")
except Exception as e:
logger.error(f"Error listing versions: {e}", exc_info=True)
return MCPErrorFormatter.from_exception(e, "list versions")
@mcp.tool()
async def get_version(
ctx: Context, project_id: str, field_name: str, version_number: int
) -> str:
"""
Get detailed information about a specific version.
Args:
project_id: Project UUID (required)
field_name: Field name - "docs", "features", "data", "prd" (required)
version_number: Version number to retrieve (required)
Returns:
JSON with complete version details and content
Example:
get_version(project_id="uuid", field_name="docs", version_number=3)
"""
try:
api_url = get_api_url()
timeout = get_default_timeout()
async with httpx.AsyncClient(timeout=timeout) as client:
response = await client.get(
urljoin(
api_url,
f"/api/projects/{project_id}/versions/{field_name}/{version_number}",
)
)
if response.status_code == 200:
result = response.json()
return json.dumps({
"success": True,
"version": result.get("version"),
"content": result.get("content"),
})
elif response.status_code == 404:
return MCPErrorFormatter.format_error(
error_type="not_found",
message=f"Version {version_number} not found for field {field_name}",
suggestion="Check that the version number and field name are correct",
http_status=404,
)
else:
return MCPErrorFormatter.from_http_error(response, "get version")
except httpx.RequestError as e:
return MCPErrorFormatter.from_exception(
e,
"get version",
{
"project_id": project_id,
"field_name": field_name,
"version_number": version_number,
},
)
except Exception as e:
logger.error(f"Error getting version: {e}", exc_info=True)
return MCPErrorFormatter.from_exception(e, "get version")
@mcp.tool()
async def restore_version(
async def manage_version(
ctx: Context,
action: str, # "create" | "restore"
project_id: str,
field_name: str,
version_number: int,
restored_by: str = "system",
version_number: int | None = None,
content: dict[str, Any] | list[dict[str, Any]] | None = None,
change_summary: str | None = None,
document_id: str | None = None,
created_by: str = "system",
) -> str:
"""
Restore a previous version.
Manage versions (consolidated: create/restore).
Args:
action: "create" | "restore"
project_id: Project UUID (required)
field_name: Field name - "docs", "features", "data", "prd" (required)
version_number: Version number to restore (required)
restored_by: Identifier of who is restoring (optional, defaults to "system")
Returns:
JSON confirmation of restoration
Example:
restore_version(project_id="uuid", field_name="docs", version_number=2)
field_name: docs/features/data/prd
version_number: Version to restore (for restore action)
content: Content to snapshot (for create action)
change_summary: What changed (for create)
document_id: Specific doc ID (optional)
created_by: Who created version
Examples:
manage_version("create", project_id="p-1", field_name="docs",
content=[...], change_summary="Updated API")
manage_version("restore", project_id="p-1", field_name="docs",
version_number=3)
Returns: {success: bool, version?: object, message: string}
"""
try:
api_url = get_api_url()
timeout = get_default_timeout()
async with httpx.AsyncClient(timeout=timeout) as client:
response = await client.post(
urljoin(
api_url,
f"/api/projects/{project_id}/versions/{field_name}/{version_number}/restore",
),
json={"restored_by": restored_by},
)
if response.status_code == 200:
result = response.json()
return json.dumps({
"success": True,
"message": result.get(
"message", f"Version {version_number} restored successfully"
),
})
elif response.status_code == 404:
return MCPErrorFormatter.format_error(
error_type="not_found",
message=f"Version {version_number} not found for field {field_name}",
suggestion="Check that the version number exists for this field",
http_status=404,
if action == "create":
if not content:
return MCPErrorFormatter.format_error(
"validation_error",
"content required for create"
)
response = await client.post(
urljoin(api_url, f"/api/projects/{project_id}/versions"),
json={
"field_name": field_name,
"content": content,
"change_summary": change_summary or "No summary provided",
"document_id": document_id,
"created_by": created_by,
}
)
if response.status_code == 200:
result = response.json()
version = result.get("version")
# Don't optimize for create - return full version
return json.dumps({
"success": True,
"version": version,
"message": result.get("message", "Version created successfully")
})
else:
return MCPErrorFormatter.from_http_error(response, "create version")
elif action == "restore":
if version_number is None:
return MCPErrorFormatter.format_error(
"validation_error",
"version_number required for restore"
)
response = await client.post(
urljoin(api_url, f"/api/projects/{project_id}/versions/{field_name}/{version_number}/restore"),
json={}
)
if response.status_code == 200:
result = response.json()
return json.dumps({
"success": True,
"message": result.get("message", "Version restored successfully"),
"field_name": field_name,
"version_number": version_number
})
else:
return MCPErrorFormatter.from_http_error(response, "restore version")
else:
return MCPErrorFormatter.from_http_error(response, "restore version")
return MCPErrorFormatter.format_error(
"invalid_action",
f"Unknown action: {action}. Use 'create' or 'restore'"
)
except httpx.RequestError as e:
return MCPErrorFormatter.from_exception(
e,
"restore version",
{
"project_id": project_id,
"field_name": field_name,
"version_number": version_number,
},
)
return MCPErrorFormatter.from_exception(e, f"{action} version")
except Exception as e:
logger.error(f"Error restoring version: {e}", exc_info=True)
return MCPErrorFormatter.from_exception(e, "restore version")
logger.error(f"Error managing version ({action}): {e}", exc_info=True)
return MCPErrorFormatter.from_exception(e, f"{action} version")

550
python/src/mcp_server/features/projects/project_tools.py
View File

@ -1,8 +1,7 @@
"""
Simple project management tools for Archon MCP Server.
Consolidated project management tools for Archon MCP Server.
Provides separate, focused tools for each project operation.
No complex PRP examples - just straightforward project management.
Reduces the number of individual CRUD operations while maintaining full functionality.
"""
import asyncio
@ -24,331 +23,308 @@ from src.server.config.service_discovery import get_api_url
logger = logging.getLogger(__name__)
# Optimization constants
MAX_DESCRIPTION_LENGTH = 1000
DEFAULT_PAGE_SIZE = 10 # Reduced from 50
def truncate_text(text: str, max_length: int = MAX_DESCRIPTION_LENGTH) -> str:
"""Truncate text to maximum length with ellipsis."""
if text and len(text) > max_length:
return text[:max_length - 3] + "..."
return text
def optimize_project_response(project: dict) -> dict:
"""Optimize project object for MCP response."""
project = project.copy() # Don't modify original
# Truncate description if present
if "description" in project and project["description"]:
project["description"] = truncate_text(project["description"])
# Remove or summarize large fields
if "features" in project and isinstance(project["features"], list):
project["features_count"] = len(project["features"])
if len(project["features"]) > 3:
project["features"] = project["features"][:3] # Keep first 3
return project
def register_project_tools(mcp: FastMCP):
"""Register individual project management tools with the MCP server."""
"""Register consolidated project management tools with the MCP server."""
@mcp.tool()
async def create_project(
async def find_projects(
ctx: Context,
title: str,
description: str = "",
github_repo: str | None = None,
project_id: str | None = None, # For getting single project
query: str | None = None, # Search capability
page: int = 1,
per_page: int = DEFAULT_PAGE_SIZE,
) -> str:
"""
Create a new project with automatic AI assistance.
The project creation starts a background process that generates PRP documentation
and initial tasks based on the title and description.
List and search projects (consolidated: list + search + get).
Args:
title: Project title - should be descriptive (required)
description: Project description explaining goals and scope
github_repo: GitHub repository URL (e.g., "https://github.com/org/repo")
project_id: Get specific project by ID (returns full details)
query: Keyword search in title/description
page: Page number for pagination
per_page: Items per page (default: 10)
Returns:
JSON with project details:
{
"success": true,
"project": {...},
"project_id": "550e8400-e29b-41d4-a716-446655440000",
"message": "Project created successfully"
}
JSON array of projects or single project (optimized payloads for lists)
Examples:
# Simple project
create_project(
title="Task Management API",
description="RESTful API for managing tasks and projects"
)
# Project with GitHub integration
create_project(
title="OAuth2 Authentication System",
description="Implement secure OAuth2 authentication with multiple providers",
github_repo="https://github.com/myorg/auth-service"
)
list_projects() # All projects
list_projects(query="auth") # Search projects
list_projects(project_id="proj-123") # Get specific project
"""
try:
api_url = get_api_url()
timeout = get_default_timeout()
async with httpx.AsyncClient(timeout=timeout) as client:
response = await client.post(
urljoin(api_url, "/api/projects"),
json={"title": title, "description": description, "github_repo": github_repo},
)
if response.status_code == 200:
result = response.json()
# Handle async project creation
if "progress_id" in result:
# Poll for completion with proper error handling and backoff
max_attempts = get_max_polling_attempts()
polling_timeout = get_polling_timeout()
for attempt in range(max_attempts):
try:
# Exponential backoff
sleep_interval = get_polling_interval(attempt)
await asyncio.sleep(sleep_interval)
# Create new client with polling timeout
async with httpx.AsyncClient(
timeout=polling_timeout
) as poll_client:
list_response = await poll_client.get(
urljoin(api_url, "/api/projects")
)
list_response.raise_for_status() # Raise on HTTP errors
response_data = list_response.json()
# Extract projects array from response
projects = response_data.get("projects", [])
# Find project with matching title created recently
for proj in projects:
if proj.get("title") == title:
return json.dumps({
"success": True,
"project": proj,
"project_id": proj["id"],
"message": f"Project created successfully with ID: {proj['id']}",
})
except httpx.RequestError as poll_error:
logger.warning(
f"Polling attempt {attempt + 1}/{max_attempts} failed: {poll_error}"
)
if attempt == max_attempts - 1: # Last attempt
return MCPErrorFormatter.format_error(
error_type="polling_timeout",
message=f"Project creation polling failed after {max_attempts} attempts",
details={
"progress_id": result["progress_id"],
"title": title,
"last_error": str(poll_error),
},
suggestion="The project may still be creating. Use list_projects to check status",
)
except Exception as poll_error:
logger.warning(
f"Unexpected error during polling attempt {attempt + 1}: {poll_error}"
)
# If we couldn't find it after polling
return json.dumps({
"success": True,
"progress_id": result["progress_id"],
"message": f"Project creation in progress after {max_attempts} checks. Use list_projects to find it once complete.",
})
# Single project get mode
if project_id:
async with httpx.AsyncClient(timeout=timeout) as client:
response = await client.get(urljoin(api_url, f"/api/projects/{project_id}"))
if response.status_code == 200:
project = response.json()
# Don't optimize single project get - return full details
return json.dumps({"success": True, "project": project})
elif response.status_code == 404:
return MCPErrorFormatter.format_error(
error_type="not_found",
message=f"Project {project_id} not found",
suggestion="Verify the project ID is correct",
http_status=404,
)
else:
# Direct response (shouldn't happen with current API)
return json.dumps({"success": True, "project": result})
else:
return MCPErrorFormatter.from_http_error(response, "create project")
except httpx.ConnectError as e:
return MCPErrorFormatter.from_exception(
e, "create project", {"title": title, "api_url": api_url}
)
except httpx.TimeoutException as e:
return MCPErrorFormatter.from_exception(
e, "create project", {"title": title, "timeout": str(timeout)}
)
except Exception as e:
logger.error(f"Error creating project: {e}", exc_info=True)
return MCPErrorFormatter.from_exception(e, "create project", {"title": title})
@mcp.tool()
async def list_projects(ctx: Context) -> str:
"""
List all projects.
Returns:
JSON array of all projects with their basic information
Example:
list_projects()
"""
try:
api_url = get_api_url()
timeout = get_default_timeout()
return MCPErrorFormatter.from_http_error(response, "get project")
# List mode
async with httpx.AsyncClient(timeout=timeout) as client:
# CRITICAL: Pass include_content=False for lightweight response
response = await client.get(
urljoin(api_url, "/api/projects"),
params={"include_content": False}
)
response = await client.get(urljoin(api_url, "/api/projects"))
if response.status_code == 200:
response_data = response.json()
# Response already includes projects array and count
data = response.json()
projects = data.get("projects", [])
# Apply search filter if provided
if query:
query_lower = query.lower()
projects = [
p for p in projects
if query_lower in p.get("title", "").lower()
or query_lower in p.get("description", "").lower()
]
# Apply pagination
start_idx = (page - 1) * per_page
end_idx = start_idx + per_page
paginated = projects[start_idx:end_idx]
# Optimize project responses
optimized = [optimize_project_response(p) for p in paginated]
return json.dumps({
"success": True,
"projects": response_data,
"count": response_data.get("count", 0),
"projects": optimized,
"count": len(optimized),
"total": len(projects),
"page": page,
"per_page": per_page,
"query": query
})
else:
return MCPErrorFormatter.from_http_error(response, "list projects")
except httpx.RequestError as e:
return MCPErrorFormatter.from_exception(e, "list projects", {"api_url": api_url})
return MCPErrorFormatter.from_exception(e, "list projects")
except Exception as e:
logger.error(f"Error listing projects: {e}", exc_info=True)
return MCPErrorFormatter.from_exception(e, "list projects")
@mcp.tool()
async def get_project(ctx: Context, project_id: str) -> str:
"""
Get detailed information about a specific project.
Args:
project_id: UUID of the project
Returns:
JSON with complete project details
Example:
get_project(project_id="550e8400-e29b-41d4-a716-446655440000")
"""
try:
api_url = get_api_url()
timeout = get_default_timeout()
async with httpx.AsyncClient(timeout=timeout) as client:
response = await client.get(urljoin(api_url, f"/api/projects/{project_id}"))
if response.status_code == 200:
project = response.json()
return json.dumps({"success": True, "project": project})
elif response.status_code == 404:
return MCPErrorFormatter.format_error(
error_type="not_found",
message=f"Project {project_id} not found",
suggestion="Verify the project ID is correct",
http_status=404,
)
else:
return MCPErrorFormatter.from_http_error(response, "get project")
except httpx.RequestError as e:
return MCPErrorFormatter.from_exception(e, "get project", {"project_id": project_id})
except Exception as e:
logger.error(f"Error getting project: {e}", exc_info=True)
return MCPErrorFormatter.from_exception(e, "get project")
@mcp.tool()
async def delete_project(ctx: Context, project_id: str) -> str:
"""
Delete a project.
Args:
project_id: UUID of the project to delete
Returns:
JSON confirmation of deletion
Example:
delete_project(project_id="550e8400-e29b-41d4-a716-446655440000")
"""
try:
api_url = get_api_url()
timeout = get_default_timeout()
async with httpx.AsyncClient(timeout=timeout) as client:
response = await client.delete(urljoin(api_url, f"/api/projects/{project_id}"))
if response.status_code == 200:
return json.dumps({
"success": True,
"message": f"Project {project_id} deleted successfully",
})
elif response.status_code == 404:
return MCPErrorFormatter.format_error(
error_type="not_found",
message=f"Project {project_id} not found",
suggestion="Verify the project ID is correct",
http_status=404,
)
else:
return MCPErrorFormatter.from_http_error(response, "delete project")
except httpx.RequestError as e:
return MCPErrorFormatter.from_exception(e, "delete project", {"project_id": project_id})
except Exception as e:
logger.error(f"Error deleting project: {e}", exc_info=True)
return MCPErrorFormatter.from_exception(e, "delete project")
@mcp.tool()
async def update_project(
async def manage_project(
ctx: Context,
project_id: str,
action: str, # "create" | "update" | "delete"
project_id: str | None = None,
title: str | None = None,
description: str | None = None,
github_repo: str | None = None,
) -> str:
"""
Update a project's basic information.
Manage projects (consolidated: create/update/delete).
Args:
project_id: UUID of the project to update
title: New title (optional)
description: New description (optional)
github_repo: New GitHub repository URL (optional)
Returns:
JSON with updated project details
Example:
update_project(project_id="550e8400-e29b-41d4-a716-446655440000",
title="Updated Project Title")
action: "create" | "update" | "delete"
project_id: Project UUID for update/delete
title: Project title (required for create)
description: Project goals and scope
github_repo: GitHub URL (e.g. "https://github.com/org/repo")
Examples:
manage_project("create", title="Auth System")
manage_project("update", project_id="p-1", description="Updated")
manage_project("delete", project_id="p-1")
Returns: {success: bool, project?: object, message: string}
"""
try:
api_url = get_api_url()
timeout = get_default_timeout()
# Build update payload with only provided fields
update_data = {}
if title is not None:
update_data["title"] = title
if description is not None:
update_data["description"] = description
if github_repo is not None:
update_data["github_repo"] = github_repo
if not update_data:
return MCPErrorFormatter.format_error(
error_type="validation_error",
message="No fields to update",
suggestion="Provide at least one field to update (title, description, or github_repo)",
)
async with httpx.AsyncClient(timeout=timeout) as client:
response = await client.put(
urljoin(api_url, f"/api/projects/{project_id}"), json=update_data
)
if response.status_code == 200:
project = response.json()
return json.dumps({
"success": True,
"project": project,
"message": "Project updated successfully",
})
elif response.status_code == 404:
return MCPErrorFormatter.format_error(
error_type="not_found",
message=f"Project {project_id} not found",
suggestion="Verify the project ID is correct",
http_status=404,
if action == "create":
if not title:
return MCPErrorFormatter.format_error(
"validation_error",
"title required for create"
)
response = await client.post(
urljoin(api_url, "/api/projects"),
json={
"title": title,
"description": description or "",
"github_repo": github_repo
}
)
if response.status_code == 200:
result = response.json()
# Handle async project creation with polling
if "progress_id" in result:
max_attempts = get_max_polling_attempts()
polling_timeout = get_polling_timeout()
for attempt in range(max_attempts):
try:
# Exponential backoff
sleep_interval = get_polling_interval(attempt)
await asyncio.sleep(sleep_interval)
async with httpx.AsyncClient(timeout=polling_timeout) as poll_client:
poll_response = await poll_client.get(
urljoin(api_url, f"/api/progress/{result['progress_id']}")
)
if poll_response.status_code == 200:
poll_data = poll_response.json()
if poll_data.get("status") == "completed":
project = poll_data.get("result", {}).get("project", {})
return json.dumps({
"success": True,
"project": optimize_project_response(project),
"project_id": project.get("id"),
"message": poll_data.get("result", {}).get("message", "Project created successfully")
})
elif poll_data.get("status") == "failed":
error_msg = poll_data.get("error", "Project creation failed")
return MCPErrorFormatter.format_error(
"creation_failed",
error_msg,
details=poll_data.get("details")
)
# Continue polling if still processing
except httpx.RequestError as poll_error:
logger.warning(f"Polling attempt {attempt + 1} failed: {poll_error}")
if attempt == max_attempts - 1:
return MCPErrorFormatter.format_error(
"timeout",
"Project creation timed out",
suggestion="Check project status manually"
)
return MCPErrorFormatter.format_error(
"timeout",
"Project creation timed out after maximum attempts",
details={"progress_id": result.get("progress_id")}
)
else:
# Synchronous response
project = result.get("project", {})
return json.dumps({
"success": True,
"project": optimize_project_response(project),
"project_id": project.get("id"),
"message": result.get("message", "Project created successfully")
})
else:
return MCPErrorFormatter.from_http_error(response, "create project")
elif action == "update":
if not project_id:
return MCPErrorFormatter.format_error(
"validation_error",
"project_id required for update"
)
update_data = {}
if title is not None:
update_data["title"] = title
if description is not None:
update_data["description"] = description
if github_repo is not None:
update_data["github_repo"] = github_repo
if not update_data:
return MCPErrorFormatter.format_error(
"validation_error",
"No fields to update"
)
response = await client.put(
urljoin(api_url, f"/api/projects/{project_id}"),
json=update_data
)
if response.status_code == 200:
result = response.json()
project = result.get("project")
if project:
project = optimize_project_response(project)
return json.dumps({
"success": True,
"project": project,
"message": result.get("message", "Project updated successfully")
})
else:
return MCPErrorFormatter.from_http_error(response, "update project")
elif action == "delete":
if not project_id:
return MCPErrorFormatter.format_error(
"validation_error",
"project_id required for delete"
)
response = await client.delete(
urljoin(api_url, f"/api/projects/{project_id}")
)
if response.status_code == 200:
result = response.json()
return json.dumps({
"success": True,
"message": result.get("message", "Project deleted successfully")
})
else:
return MCPErrorFormatter.from_http_error(response, "delete project")
else:
return MCPErrorFormatter.from_http_error(response, "update project")
return MCPErrorFormatter.format_error(
"invalid_action",
f"Unknown action: {action}"
)
except httpx.RequestError as e:
return MCPErrorFormatter.from_exception(e, "update project", {"project_id": project_id})
return MCPErrorFormatter.from_exception(e, f"{action} project")
except Exception as e:
logger.error(f"Error updating project: {e}", exc_info=True)
return MCPErrorFormatter.from_exception(e, "update project")
logger.error(f"Error managing project ({action}): {e}", exc_info=True)
return MCPErrorFormatter.from_exception(e, f"{action} project")

12
python/src/mcp_server/features/rag/__init__.py Normal file
View File

@ -0,0 +1,12 @@
"""
RAG (Retrieval-Augmented Generation) tools for Archon MCP Server.
This module provides tools for knowledge base operations:
- perform_rag_query: Search knowledge base for relevant content
- search_code_examples: Find code examples in the knowledge base
- get_available_sources: List available knowledge sources
"""
from .rag_tools import register_rag_tools
__all__ = ["register_rag_tools"]

6
python/src/mcp_server/modules/rag_module.py → python/src/mcp_server/features/rag/rag_tools.py
View File

@ -40,7 +40,7 @@ def register_rag_tools(mcp: FastMCP):
"""Register all RAG tools with the MCP server."""
@mcp.tool()
async def get_available_sources(ctx: Context) -> str:
async def rag_get_available_sources(ctx: Context) -> str:
"""
Get list of available sources in the knowledge base.
@ -77,7 +77,7 @@ def register_rag_tools(mcp: FastMCP):
return json.dumps({"success": False, "error": str(e)}, indent=2)
@mcp.tool()
async def perform_rag_query(
async def rag_search_knowledge_base(
ctx: Context, query: str, source_domain: str | None = None, match_count: int = 5
) -> str:
"""
@ -134,7 +134,7 @@ def register_rag_tools(mcp: FastMCP):
return json.dumps({"success": False, "results": [], "error": str(e)}, indent=2)
@mcp.tool()
async def search_code_examples(
async def rag_search_code_examples(
ctx: Context, query: str, source_domain: str | None = None, match_count: int = 5
) -> str:
"""

569
python/src/mcp_server/features/tasks/task_tools.py
View File

@ -1,8 +1,7 @@
"""
Simple task management tools for Archon MCP Server.
Consolidated task management tools for Archon MCP Server.
Provides separate, focused tools for each task operation.
Mirrors the functionality of the original manage_task tool but with individual tools.
Reduces the number of individual CRUD operations while maintaining full functionality.
"""
import json
@ -19,175 +18,107 @@ from src.server.config.service_discovery import get_api_url
logger = logging.getLogger(__name__)
# Optimization constants
MAX_DESCRIPTION_LENGTH = 1000
DEFAULT_PAGE_SIZE = 10 # Reduced from 50
def truncate_text(text: str, max_length: int = MAX_DESCRIPTION_LENGTH) -> str:
"""Truncate text to maximum length with ellipsis."""
if text and len(text) > max_length:
return text[:max_length - 3] + "..."
return text
def optimize_task_response(task: dict) -> dict:
"""Optimize task object for MCP response."""
task = task.copy() # Don't modify original
# Truncate description if present
if "description" in task and task["description"]:
task["description"] = truncate_text(task["description"])
# Replace arrays with counts
if "sources" in task and isinstance(task["sources"], list):
task["sources_count"] = len(task["sources"])
del task["sources"]
if "code_examples" in task and isinstance(task["code_examples"], list):
task["code_examples_count"] = len(task["code_examples"])
del task["code_examples"]
return task
def register_task_tools(mcp: FastMCP):
"""Register individual task management tools with the MCP server."""
"""Register consolidated task management tools with the MCP server."""
@mcp.tool()
async def create_task(
ctx: Context,
project_id: str,
title: str,
description: str = "",
assignee: str = "User",
task_order: int = 0,
feature: str | None = None,
sources: list[dict[str, str]] | None = None,
code_examples: list[dict[str, str]] | None = None,
) -> str:
"""
Create a new task in a project.
Args:
project_id: Project UUID (required)
title: Task title - should be specific and actionable (required)
description: Detailed task description with acceptance criteria
assignee: Who will work on this task. Options:
- "User": For manual tasks
- "Archon": For AI-driven tasks
- "AI IDE Agent": For code implementation
- "prp-executor": For PRP coordination
- "prp-validator": For testing/validation
task_order: Priority within status (0-100, higher = more priority)
feature: Feature label for grouping related tasks (e.g., "authentication")
sources: List of source references. Each source should have:
- "url": Link to documentation or file path
- "type": Type of source (e.g., "documentation", "api_spec")
- "relevance": Why this source is relevant
code_examples: List of code examples. Each example should have:
- "file": Path to the file
- "function": Function or class name
- "purpose": Why this example is relevant
Returns:
JSON with task details including task_id:
{
"success": true,
"task": {...},
"task_id": "task-123",
"message": "Task created successfully"
}
Examples:
# Simple task
create_task(
project_id="550e8400-e29b-41d4-a716-446655440000",
title="Add user authentication",
description="Implement JWT-based authentication with refresh tokens"
)
# Task with sources and examples
create_task(
project_id="550e8400-e29b-41d4-a716-446655440000",
title="Implement OAuth2 Google provider",
description="Add Google OAuth2 with PKCE security",
assignee="AI IDE Agent",
task_order=10,
feature="authentication",
sources=[
{
"url": "https://developers.google.com/identity/protocols/oauth2",
"type": "documentation",
"relevance": "Official OAuth2 implementation guide"
},
{
"url": "docs/auth/README.md",
"type": "internal_docs",
"relevance": "Current auth architecture"
}
],
code_examples=[
{
"file": "src/auth/base.py",
"function": "BaseAuthProvider",
"purpose": "Base class to extend"
},
{
"file": "tests/auth/test_oauth.py",
"function": "test_oauth_flow",
"purpose": "Test pattern to follow"
}
]
)
"""
try:
api_url = get_api_url()
timeout = get_default_timeout()
async with httpx.AsyncClient(timeout=timeout) as client:
response = await client.post(
urljoin(api_url, "/api/tasks"),
json={
"project_id": project_id,
"title": title,
"description": description,
"assignee": assignee,
"task_order": task_order,
"feature": feature,
"sources": sources or [],
"code_examples": code_examples or [],
},
)
if response.status_code == 200:
result = response.json()
return json.dumps({
"success": True,
"task": result.get("task"),
"task_id": result.get("task", {}).get("id"),
"message": result.get("message", "Task created successfully"),
})
else:
return MCPErrorFormatter.from_http_error(response, "create task")
except httpx.RequestError as e:
return MCPErrorFormatter.from_exception(
e, "create task", {"project_id": project_id, "title": title}
)
except Exception as e:
logger.error(f"Error creating task: {e}", exc_info=True)
return MCPErrorFormatter.from_exception(e, "create task")
@mcp.tool()
async def list_tasks(
async def find_tasks(
ctx: Context,
query: str | None = None, # Add search capability
task_id: str | None = None, # For getting single task
filter_by: str | None = None,
filter_value: str | None = None,
project_id: str | None = None,
include_closed: bool = False,
include_closed: bool = True,
page: int = 1,
per_page: int = 50,
per_page: int = DEFAULT_PAGE_SIZE, # Use optimized default
) -> str:
"""
List tasks with filtering options.
Find and search tasks (consolidated: list + search + get).
Args:
query: Keyword search in title, description, feature (optional)
task_id: Get specific task by ID (returns full details)
filter_by: "status" | "project" | "assignee" (optional)
filter_value: Filter value (e.g., "todo", "doing", "review", "done")
project_id: Project UUID (optional, for additional filtering)
include_closed: Include done tasks in results
page: Page number for pagination
per_page: Items per page
per_page: Items per page (default: 10)
Returns:
JSON array of tasks with pagination info
JSON array of tasks or single task (optimized payloads for lists)
Examples:
list_tasks() # All tasks
list_tasks(filter_by="status", filter_value="todo") # Only todo tasks
list_tasks(filter_by="project", filter_value="project-uuid") # Tasks for specific project
find_tasks() # All tasks
find_tasks(query="auth") # Search for "auth"
find_tasks(task_id="task-123") # Get specific task (full details)
find_tasks(filter_by="status", filter_value="todo") # Only todo tasks
"""
try:
api_url = get_api_url()
timeout = get_default_timeout()
# Build URL and parameters based on filter type
# Single task get mode
if task_id:
async with httpx.AsyncClient(timeout=timeout) as client:
response = await client.get(urljoin(api_url, f"/api/tasks/{task_id}"))
if response.status_code == 200:
task = response.json()
# Don't optimize single task get - return full details
return json.dumps({"success": True, "task": task})
elif response.status_code == 404:
return MCPErrorFormatter.format_error(
error_type="not_found",
message=f"Task {task_id} not found",
suggestion="Verify the task ID is correct",
http_status=404,
)
else:
return MCPErrorFormatter.from_http_error(response, "get task")
# List mode with search and filters
params: dict[str, Any] = {
"page": page,
"per_page": per_page,
"exclude_large_fields": True, # Always exclude large fields in MCP responses
}
# Add search query if provided
if query:
params["q"] = query
if filter_by == "project" and filter_value:
# Use project-specific endpoint for project filtering
url = urljoin(api_url, f"/api/projects/{filter_value}/tasks")
@ -199,57 +130,57 @@ def register_task_tools(mcp: FastMCP):
params["include_closed"] = include_closed
if project_id:
params["project_id"] = project_id
elif project_id:
# Direct project_id parameter provided
url = urljoin(api_url, "/api/tasks")
params["project_id"] = project_id
params["include_closed"] = include_closed
else:
# Default to generic tasks endpoint
# No specific filters - get all tasks
url = urljoin(api_url, "/api/tasks")
params["include_closed"] = include_closed
if project_id:
params["project_id"] = project_id
async with httpx.AsyncClient(timeout=timeout) as client:
response = await client.get(url, params=params)
response.raise_for_status()
result = response.json()
# Normalize response format - handle both array and object responses
# Normalize response format
if isinstance(result, list):
# Direct array response
tasks = result
total_count = len(result)
elif isinstance(result, dict):
# Object response - check for standard fields
if "tasks" in result:
tasks = result["tasks"]
total_count = result.get("total_count", len(tasks))
elif "data" in result:
# Alternative format with 'data' field
tasks = result["data"]
total_count = result.get("total", len(tasks))
else:
# Unknown object format
return MCPErrorFormatter.format_error(
error_type="invalid_response",
message="Unexpected response format from API",
details={"response_keys": list(result.keys())},
suggestion="The API response format may have changed. Please check for updates.",
)
else:
# Completely unexpected format
return MCPErrorFormatter.format_error(
error_type="invalid_response",
message="Invalid response type from API",
details={"response_type": type(result).__name__},
suggestion="Expected list or object, got different type.",
)
# Optimize task responses
optimized_tasks = [optimize_task_response(task) for task in tasks]
return json.dumps({
"success": True,
"tasks": tasks,
"tasks": optimized_tasks,
"total_count": total_count,
"count": len(tasks),
"count": len(optimized_tasks),
"query": query, # Include search query in response
})
except httpx.RequestError as e:
return MCPErrorFormatter.from_exception(
e, "list tasks", {"filter_by": filter_by, "filter_value": filter_value}
@ -259,194 +190,166 @@ def register_task_tools(mcp: FastMCP):
return MCPErrorFormatter.from_exception(e, "list tasks")
@mcp.tool()
async def get_task(ctx: Context, task_id: str) -> str:
"""
Get detailed information about a specific task.
Args:
task_id: UUID of the task
Returns:
JSON with complete task details
Example:
get_task(task_id="task-uuid")
"""
try:
api_url = get_api_url()
timeout = get_default_timeout()
async with httpx.AsyncClient(timeout=timeout) as client:
response = await client.get(urljoin(api_url, f"/api/tasks/{task_id}"))
if response.status_code == 200:
task = response.json()
return json.dumps({"success": True, "task": task})
elif response.status_code == 404:
return MCPErrorFormatter.format_error(
error_type="not_found",
message=f"Task {task_id} not found",
suggestion="Verify the task ID is correct",
http_status=404,
)
else:
return MCPErrorFormatter.from_http_error(response, "get task")
except httpx.RequestError as e:
return MCPErrorFormatter.from_exception(e, "get task", {"task_id": task_id})
except Exception as e:
logger.error(f"Error getting task: {e}", exc_info=True)
return MCPErrorFormatter.from_exception(e, "get task")
@mcp.tool()
async def update_task(
async def manage_task(
ctx: Context,
task_id: str,
action: str, # "create" | "update" | "delete"
task_id: str | None = None,
project_id: str | None = None,
title: str | None = None,
description: str | None = None,
status: str | None = None,
assignee: str | None = None,
task_order: int | None = None,
feature: str | None = None,
sources: list[dict[str, str]] | None = None,
code_examples: list[dict[str, str]] | None = None,
feature: str | None = None
) -> str:
"""
Update a task's properties.
Manage tasks (consolidated: create/update/delete).
Args:
task_id: UUID of the task to update
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
action: "create" | "update" | "delete"
task_id: Task UUID for update/delete
project_id: Project UUID for create
title: Task title text
description: Detailed task description
status: "todo" | "doing" | "review" | "done"
assignee: "User" | "Archon" | "AI IDE Agent"
task_order: Priority 0-100 (higher = more priority)
feature: Feature label for grouping
Examples:
update_task(task_id="uuid", status="doing")
update_task(task_id="uuid", title="New Title", description="Updated description")
manage_task("create", project_id="p-1", title="Fix auth bug")
manage_task("update", task_id="t-1", status="doing")
manage_task("delete", task_id="t-1")
Returns: {success: bool, task?: object, message: string}
"""
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
)
if response.status_code == 200:
result = response.json()
return json.dumps({
"success": True,
"task": result.get("task"),
"message": result.get("message", "Task updated successfully"),
})
if action == "create":
if not project_id or not title:
return MCPErrorFormatter.format_error(
"validation_error",
"project_id and title required for create",
suggestion="Provide both project_id and title"
)
response = await client.post(
urljoin(api_url, "/api/tasks"),
json={
"project_id": project_id,
"title": title,
"description": description or "",
"assignee": assignee or "User",
"task_order": task_order or 0,
"feature": feature,
"sources": [],
"code_examples": [],
},
)
if response.status_code == 200:
result = response.json()
task = result.get("task")
# Optimize task response
if task:
task = optimize_task_response(task)
return json.dumps({
"success": True,
"task": task,
"task_id": task.get("id") if task else None,
"message": result.get("message", "Task created successfully"),
})
else:
return MCPErrorFormatter.from_http_error(response, "create task")
elif action == "update":
if not task_id:
return MCPErrorFormatter.format_error(
"validation_error",
"task_id required for update",
suggestion="Provide task_id to update"
)
# Build update fields
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 not update_fields:
return MCPErrorFormatter.format_error(
error_type="validation_error",
message="No fields to update",
suggestion="Provide at least one field to update",
)
response = await client.put(
urljoin(api_url, f"/api/tasks/{task_id}"),
json=update_fields
)
if response.status_code == 200:
result = response.json()
task = result.get("task")
# Optimize task response
if task:
task = optimize_task_response(task)
return json.dumps({
"success": True,
"task": task,
"message": result.get("message", "Task updated successfully"),
})
else:
return MCPErrorFormatter.from_http_error(response, "update task")
elif action == "delete":
if not task_id:
return MCPErrorFormatter.format_error(
"validation_error",
"task_id required for delete",
suggestion="Provide task_id to delete"
)
response = await client.delete(
urljoin(api_url, f"/api/tasks/{task_id}")
)
if response.status_code == 200:
result = response.json()
return json.dumps({
"success": True,
"message": result.get("message", "Task deleted successfully"),
})
else:
return MCPErrorFormatter.from_http_error(response, "delete task")
else:
return MCPErrorFormatter.from_http_error(response, "update task")
return MCPErrorFormatter.format_error(
"invalid_action",
f"Unknown action: {action}",
suggestion="Use 'create', 'update', or 'delete'"
)
except httpx.RequestError as e:
return MCPErrorFormatter.from_exception(
e, "update task", {"task_id": task_id, "update_fields": list(update_fields.keys())}
e, f"{action} task", {"task_id": task_id, "project_id": project_id}
)
except Exception as e:
logger.error(f"Error updating task: {e}", exc_info=True)
return MCPErrorFormatter.from_exception(e, "update task")
@mcp.tool()
async def delete_task(ctx: Context, task_id: str) -> str:
"""
Delete/archive a task.
This removes the task from active lists but preserves it in the database
for audit purposes (soft delete).
Args:
task_id: UUID of the task to delete/archive
Returns:
JSON confirmation of deletion:
{
"success": true,
"message": "Task deleted successfully",
"subtasks_archived": 0
}
Example:
delete_task(task_id="task-123e4567-e89b-12d3-a456-426614174000")
"""
try:
api_url = get_api_url()
timeout = get_default_timeout()
async with httpx.AsyncClient(timeout=timeout) as client:
response = await client.delete(urljoin(api_url, f"/api/tasks/{task_id}"))
if response.status_code == 200:
result = response.json()
return json.dumps({
"success": True,
"message": result.get("message", f"Task {task_id} deleted successfully"),
"subtasks_archived": result.get("subtasks_archived", 0),
})
elif response.status_code == 404:
return json.dumps({
"success": False,
"error": f"Task {task_id} not found. Use list_tasks to find valid task IDs.",
})
elif response.status_code == 400:
# More specific error for bad requests
error_text = response.text
if "already archived" in error_text.lower():
return MCPErrorFormatter.format_error(
error_type="already_archived",
message=f"Task {task_id} is already archived",
suggestion="No further action needed - task is already archived",
http_status=400,
)
return MCPErrorFormatter.format_error(
error_type="validation_error",
message=f"Cannot delete task: {error_text}",
suggestion="Check if the task meets deletion requirements",
http_status=400,
)
else:
return MCPErrorFormatter.from_http_error(response, "delete task")
except httpx.RequestError as e:
return MCPErrorFormatter.from_exception(e, "delete task", {"task_id": task_id})
except Exception as e:
logger.error(f"Error deleting task: {e}", exc_info=True)
return MCPErrorFormatter.from_exception(e, "delete task")
logger.error(f"Error managing task ({action}): {e}", exc_info=True)
return MCPErrorFormatter.from_exception(e, f"{action} task")

88
python/src/mcp_server/mcp_server.py
View File

@ -197,48 +197,56 @@ MCP_INSTRUCTIONS = """
- 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
2. **Research First**: Before implementing, use rag_search_knowledge_base and rag_search_code_examples
3. **Task-Driven Development**: Never code without checking current tasks first
## 📋 Core Workflow
### 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")`
1. **Get current task**: `list_tasks(task_id="...")`
2. **Search/List tasks**: `list_tasks(query="auth", filter_by="status", filter_value="todo")`
3. **Mark as doing**: `manage_task("update", task_id="...", status="doing")`
4. **Research phase**:
- `rag_search_knowledge_base(query="...", match_count=5)`
- `rag_search_code_examples(query="...", match_count=3)`
5. **Implementation**: Code based on research findings
6. **Mark for review**: `manage_task("update", task_id="...", status="review")`
7. **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)`
### Consolidated Task Tools (Optimized ~2 tools from 5)
- `list_tasks(query=None, task_id=None, filter_by=None, filter_value=None, per_page=10)`
- **Consolidated**: list + search + get in one tool
- **NEW**: Search with keyword query parameter
- **NEW**: task_id parameter for getting single task (full details)
- Filter by status, project, or assignee
- **Optimized**: Returns truncated descriptions and array counts (lists only)
- **Default**: 10 items per page (was 50)
- `manage_task(action, task_id=None, project_id=None, ...)`
- **Consolidated**: create + update + delete in one tool
- action: "create" | "update" | "delete"
- Examples:
- `manage_task("create", project_id="p-1", title="Fix auth")`
- `manage_task("update", task_id="t-1", status="doing")`
- `manage_task("delete", task_id="t-1")`
## 🏗️ 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)`
### Project Tools (Consolidated)
- `list_projects(project_id=None, query=None, page=1, per_page=10)`
- List all projects, search by query, or get specific project by ID
- `manage_project(action, project_id=None, title=None, description=None, github_repo=None)`
- Actions: "create", "update", "delete"
### 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)`
### Document Tools (Consolidated)
- `list_documents(project_id, document_id=None, query=None, document_type=None, page=1, per_page=10)`
- List project documents, search, filter by type, or get specific document
- `manage_document(action, project_id, document_id=None, title=None, document_type=None, content=None, ...)`
- Actions: "create", "update", "delete"
## 🔍 Research Patterns
- **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()`
- **Architecture patterns**: `rag_search_knowledge_base(query="[tech] architecture patterns", match_count=5)`
- **Code examples**: `rag_search_code_examples(query="[feature] implementation", match_count=3)`
- **Source discovery**: `rag_get_available_sources()`
- Keep match_count around 3-5 for focused results
## 📊 Task Status Flow
@ -247,12 +255,12 @@ MCP_INSTRUCTIONS = """
- Use 'review' for completed work awaiting validation
- 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"
## 💾 Version Management (Consolidated)
- `list_versions(project_id, field_name=None, version_number=None, page=1, per_page=10)`
- List all versions, filter by field, or get specific version
- `manage_version(action, project_id, field_name, version_number=None, content=None, change_summary=None, ...)`
- Actions: "create", "restore"
- Field names: "docs", "features", "data", "prd"
## 🎯 Best Practices
1. **Atomic Tasks**: Create tasks that take 1-4 hours
@ -260,6 +268,12 @@ MCP_INSTRUCTIONS = """
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
## 📊 Optimization Updates
- **Payload Optimization**: Tasks in lists return truncated descriptions (200 chars)
- **Array Counts**: Source/example arrays replaced with counts in list responses
- **Smart Defaults**: Default page size reduced from 50 to 10 items
- **Search Support**: New `query` parameter in list_tasks for keyword search
"""
# Initialize the main FastMCP server with fixed configuration
@ -380,7 +394,7 @@ def register_modules():
# Import and register RAG module (HTTP-based version)
try:
from src.mcp_server.modules.rag_module import register_rag_tools
from src.mcp_server.features.rag import register_rag_tools
register_rag_tools(mcp)
modules_registered += 1

0
python/src/mcp_server/modules/models.py → python/src/mcp_server/models.py
View File

8
python/src/mcp_server/modules/__init__.py
View File

@ -1,8 +0,0 @@
"""
Modular MCP Tools Package
This package contains modular MCP tool implementations:
- rag_module: RAG and web crawling tools
- tasks_module: Task and project management tools
- ui_module: UI and interface tools (future)
"""

10
python/src/server/api_routes/projects_api.py
View File

@ -659,15 +659,16 @@ async def create_task(request: CreateTaskRequest):
async def list_tasks(
status: str | None = None,
project_id: str | None = None,
include_closed: bool = False,
include_closed: bool = True,
page: int = 1,
per_page: int = 50,
per_page: int = 10,
exclude_large_fields: bool = False,
q: str | None = None, # Search query parameter
):
"""List tasks with optional filters including status and project."""
"""List tasks with optional filters including status, project, and keyword search."""
try:
logfire.info(
f"Listing tasks | status={status} | project_id={project_id} | include_closed={include_closed} | page={page} | per_page={per_page}"
f"Listing tasks | status={status} | project_id={project_id} | include_closed={include_closed} | page={page} | per_page={per_page} | q={q}"
)
# Use TaskService to list tasks
@ -677,6 +678,7 @@ async def list_tasks(
status=status,
include_closed=include_closed,
exclude_large_fields=exclude_large_fields,
search_query=q, # Pass search query to service
)
if not success:

31
python/src/server/services/projects/task_service.py
View File

@ -144,7 +144,8 @@ class TaskService:
status: str = None,
include_closed: bool = False,
exclude_large_fields: bool = False,
include_archived: bool = False
include_archived: bool = False,
search_query: str = None
) -> tuple[bool, dict[str, Any]]:
"""
List tasks with various filters.
@ -155,6 +156,7 @@ class TaskService:
include_closed: Include done tasks
exclude_large_fields: If True, excludes sources and code_examples fields
include_archived: If True, includes archived tasks
search_query: Keyword search in title, description, and feature fields
Returns:
Tuple of (success, result_dict)
@ -194,6 +196,33 @@ class TaskService:
query = query.neq("status", "done")
filters_applied.append("exclude done tasks")
# Apply keyword search if provided
if search_query:
# Split search query into terms
search_terms = search_query.lower().split()
# Build the filter expression for AND-of-ORs
# Each term must match in at least one field (OR), and all terms must match (AND)
if len(search_terms) == 1:
# Single term: simple OR across fields
term = search_terms[0]
query = query.or_(
f"title.ilike.%{term}%,"
f"description.ilike.%{term}%,"
f"feature.ilike.%{term}%"
)
else:
# Multiple terms: use text search for proper AND logic
# Note: This requires full-text search columns to be set up in the database
# For now, we'll search for the full phrase in any field
full_query = search_query.lower()
query = query.or_(
f"title.ilike.%{full_query}%,"
f"description.ilike.%{full_query}%,"
f"feature.ilike.%{full_query}%"
)
filters_applied.append(f"search={search_query}")
# Filter out archived tasks only if not including them
if not include_archived:
query = query.or_("archived.is.null,archived.is.false")

43
python/tests/mcp_server/features/documents/test_document_tools.py
View File

@ -39,9 +39,9 @@ async def test_create_document_success(mock_mcp, mock_context):
# Register tools with mock MCP
register_document_tools(mock_mcp)
# Get the create_document function from registered tools
create_document = mock_mcp._tools.get("create_document")
assert create_document is not None, "create_document tool not registered"
# Get the manage_document function from registered tools
manage_document = mock_mcp._tools.get("manage_document")
assert manage_document is not None, "manage_document tool not registered"
# Mock HTTP response
mock_response = MagicMock()
@ -57,8 +57,9 @@ async def test_create_document_success(mock_mcp, mock_context):
mock_client.return_value.__aenter__.return_value = mock_async_client
# Test the function
result = await create_document(
result = await manage_document(
mock_context,
action="create",
project_id="project-123",
title="Test Document",
document_type="spec",
@ -72,13 +73,13 @@ async def test_create_document_success(mock_mcp, mock_context):
@pytest.mark.asyncio
async def test_list_documents_success(mock_mcp, mock_context):
async def test_find_documents_success(mock_mcp, mock_context):
"""Test successful document listing."""
register_document_tools(mock_mcp)
# Get the list_documents function from registered tools
list_documents = mock_mcp._tools.get("list_documents")
assert list_documents is not None, "list_documents tool not registered"
# Get the find_documents function from registered tools
find_documents = mock_mcp._tools.get("find_documents")
assert find_documents is not None, "find_documents tool not registered"
# Mock HTTP response
mock_response = MagicMock()
@ -95,7 +96,7 @@ async def test_list_documents_success(mock_mcp, mock_context):
mock_async_client.get.return_value = mock_response
mock_client.return_value.__aenter__.return_value = mock_async_client
result = await list_documents(mock_context, project_id="project-123")
result = await find_documents(mock_context, project_id="project-123")
result_data = json.loads(result)
assert result_data["success"] is True
@ -108,9 +109,9 @@ async def test_update_document_partial_update(mock_mcp, mock_context):
"""Test partial document update."""
register_document_tools(mock_mcp)
# Get the update_document function from registered tools
update_document = mock_mcp._tools.get("update_document")
assert update_document is not None, "update_document tool not registered"
# Get the manage_document function from registered tools
manage_document = mock_mcp._tools.get("manage_document")
assert manage_document is not None, "manage_document tool not registered"
# Mock HTTP response
mock_response = MagicMock()
@ -126,8 +127,8 @@ async def test_update_document_partial_update(mock_mcp, mock_context):
mock_client.return_value.__aenter__.return_value = mock_async_client
# Update only title
result = await update_document(
mock_context, project_id="project-123", doc_id="doc-123", title="Updated Title"
result = await manage_document(
mock_context, action="update", project_id="project-123", document_id="doc-123", title="Updated Title"
)
result_data = json.loads(result)
@ -145,9 +146,9 @@ async def test_delete_document_not_found(mock_mcp, mock_context):
"""Test deleting a non-existent document."""
register_document_tools(mock_mcp)
# Get the delete_document function from registered tools
delete_document = mock_mcp._tools.get("delete_document")
assert delete_document is not None, "delete_document tool not registered"
# Get the manage_document function from registered tools
manage_document = mock_mcp._tools.get("manage_document")
assert manage_document is not None, "manage_document tool not registered"
# Mock 404 response
mock_response = MagicMock()
@ -159,8 +160,8 @@ async def test_delete_document_not_found(mock_mcp, mock_context):
mock_async_client.delete.return_value = mock_response
mock_client.return_value.__aenter__.return_value = mock_async_client
result = await delete_document(
mock_context, project_id="project-123", doc_id="non-existent"
result = await manage_document(
mock_context, action="delete", project_id="project-123", document_id="non-existent"
)
result_data = json.loads(result)
@ -170,5 +171,5 @@ async def test_delete_document_not_found(mock_mcp, mock_context):
assert isinstance(result_data["error"], dict), (
"Error should be structured format, not string"
)
assert result_data["error"]["type"] == "not_found"
assert "not found" in result_data["error"]["message"].lower()
assert result_data["error"]["type"] == "http_error"
assert "404" in result_data["error"]["message"].lower()

43
python/tests/mcp_server/features/documents/test_version_tools.py
View File

@ -38,10 +38,10 @@ async def test_create_version_success(mock_mcp, mock_context):
"""Test successful version creation."""
register_version_tools(mock_mcp)
# Get the create_version function
create_version = mock_mcp._tools.get("create_version")
# Get the manage_version function
manage_version = mock_mcp._tools.get("manage_version")
assert create_version is not None, "create_version tool not registered"
assert manage_version is not None, "manage_version tool not registered"
# Mock HTTP response
mock_response = MagicMock()
@ -56,8 +56,9 @@ async def test_create_version_success(mock_mcp, mock_context):
mock_async_client.post.return_value = mock_response
mock_client.return_value.__aenter__.return_value = mock_async_client
result = await create_version(
result = await manage_version(
mock_context,
action="create",
project_id="project-123",
field_name="docs",
content=[{"id": "doc-1", "title": "Test Doc"}],
@ -66,8 +67,8 @@ async def test_create_version_success(mock_mcp, mock_context):
result_data = json.loads(result)
assert result_data["success"] is True
assert result_data["version_number"] == 3
assert "Version 3 created successfully" in result_data["message"]
assert result_data["version"]["version_number"] == 3
assert "Version created successfully" in result_data["message"]
@pytest.mark.asyncio
@ -75,7 +76,7 @@ async def test_create_version_invalid_field(mock_mcp, mock_context):
"""Test version creation with invalid field name."""
register_version_tools(mock_mcp)
create_version = mock_mcp._tools.get("create_version")
manage_version = mock_mcp._tools.get("manage_version")
# Mock 400 response for invalid field
mock_response = MagicMock()
@ -87,8 +88,8 @@ async def test_create_version_invalid_field(mock_mcp, mock_context):
mock_async_client.post.return_value = mock_response
mock_client.return_value.__aenter__.return_value = mock_async_client
result = await create_version(
mock_context, project_id="project-123", field_name="invalid", content={"test": "data"}
result = await manage_version(
mock_context, action="create", project_id="project-123", field_name="invalid", content={"test": "data"}
)
result_data = json.loads(result)
@ -98,7 +99,7 @@ async def test_create_version_invalid_field(mock_mcp, mock_context):
assert isinstance(result_data["error"], dict), (
"Error should be structured format, not string"
)
assert result_data["error"]["type"] == "validation_error"
assert result_data["error"]["type"] == "http_error"
@pytest.mark.asyncio
@ -106,10 +107,10 @@ async def test_restore_version_success(mock_mcp, mock_context):
"""Test successful version restoration."""
register_version_tools(mock_mcp)
# Get the restore_version function
restore_version = mock_mcp._tools.get("restore_version")
# Get the manage_version function
manage_version = mock_mcp._tools.get("manage_version")
assert restore_version is not None, "restore_version tool not registered"
assert manage_version is not None, "manage_version tool not registered"
# Mock HTTP response
mock_response = MagicMock()
@ -121,28 +122,28 @@ async def test_restore_version_success(mock_mcp, mock_context):
mock_async_client.post.return_value = mock_response
mock_client.return_value.__aenter__.return_value = mock_async_client
result = await restore_version(
result = await manage_version(
mock_context,
action="restore",
project_id="project-123",
field_name="docs",
version_number=2,
restored_by="test-user",
)
result_data = json.loads(result)
assert result_data["success"] is True
assert "Version 2 restored successfully" in result_data["message"]
assert "restored successfully" in result_data["message"]
@pytest.mark.asyncio
async def test_list_versions_with_filter(mock_mcp, mock_context):
async def test_find_versions_with_filter(mock_mcp, mock_context):
"""Test listing versions with field name filter."""
register_version_tools(mock_mcp)
# Get the list_versions function
list_versions = mock_mcp._tools.get("list_versions")
# Get the find_versions function
find_versions = mock_mcp._tools.get("find_versions")
assert list_versions is not None, "list_versions tool not registered"
assert find_versions is not None, "find_versions tool not registered"
# Mock HTTP response
mock_response = MagicMock()
@ -159,7 +160,7 @@ async def test_list_versions_with_filter(mock_mcp, mock_context):
mock_async_client.get.return_value = mock_response
mock_client.return_value.__aenter__.return_value = mock_async_client
result = await list_versions(mock_context, project_id="project-123", field_name="docs")
result = await find_versions(mock_context, project_id="project-123", field_name="docs")
result_data = json.loads(result)
assert result_data["success"] is True

50
python/tests/mcp_server/features/projects/test_project_tools.py
View File

@ -39,10 +39,10 @@ async def test_create_project_success(mock_mcp, mock_context):
"""Test successful project creation with polling."""
register_project_tools(mock_mcp)
# Get the create_project function
create_project = mock_mcp._tools.get("create_project")
# Get the manage_project function
manage_project = mock_mcp._tools.get("manage_project")
assert create_project is not None, "create_project tool not registered"
assert manage_project is not None, "manage_project tool not registered"
# Mock initial creation response with progress_id
mock_create_response = MagicMock()
@ -52,27 +52,29 @@ async def test_create_project_success(mock_mcp, mock_context):
"message": "Project creation started",
}
# Mock list projects response for polling - API returns dict with projects array
mock_list_response = MagicMock()
mock_list_response.status_code = 200
mock_list_response.json.return_value = {
"projects": [
{"id": "project-123", "title": "Test Project", "created_at": "2024-01-01"}
],
"count": 1
# Mock progress endpoint response for polling
mock_progress_response = MagicMock()
mock_progress_response.status_code = 200
mock_progress_response.json.return_value = {
"status": "completed",
"result": {
"project": {"id": "project-123", "title": "Test Project", "created_at": "2024-01-01"},
"message": "Project created successfully"
}
}
with patch("src.mcp_server.features.projects.project_tools.httpx.AsyncClient") as mock_client:
mock_async_client = AsyncMock()
# First call creates project, subsequent calls list projects
mock_async_client.post.return_value = mock_create_response
mock_async_client.get.return_value = mock_list_response
mock_async_client.get.return_value = mock_progress_response
mock_client.return_value.__aenter__.return_value = mock_async_client
# Mock sleep to speed up test
with patch("asyncio.sleep", new_callable=AsyncMock):
result = await create_project(
result = await manage_project(
mock_context,
action="create",
title="Test Project",
description="A test project",
github_repo="https://github.com/test/repo",
@ -90,7 +92,7 @@ async def test_create_project_direct_response(mock_mcp, mock_context):
"""Test project creation with direct response (no polling)."""
register_project_tools(mock_mcp)
create_project = mock_mcp._tools.get("create_project")
manage_project = mock_mcp._tools.get("manage_project")
# Mock direct creation response (no progress_id)
mock_create_response = MagicMock()
@ -105,7 +107,7 @@ async def test_create_project_direct_response(mock_mcp, mock_context):
mock_async_client.post.return_value = mock_create_response
mock_client.return_value.__aenter__.return_value = mock_async_client
result = await create_project(mock_context, title="Test Project")
result = await manage_project(mock_context, action="create", title="Test Project")
result_data = json.loads(result)
assert result_data["success"] is True
@ -114,14 +116,14 @@ async def test_create_project_direct_response(mock_mcp, mock_context):
@pytest.mark.asyncio
async def test_list_projects_success(mock_mcp, mock_context):
async def test_find_projects_success(mock_mcp, mock_context):
"""Test listing projects."""
register_project_tools(mock_mcp)
# Get the list_projects function
list_projects = mock_mcp._tools.get("list_projects")
# Get the find_projects function
find_projects = mock_mcp._tools.get("find_projects")
assert list_projects is not None, "list_projects tool not registered"
assert find_projects is not None, "find_projects tool not registered"
# Mock HTTP response - API returns dict with projects array
mock_response = MagicMock()
@ -139,7 +141,7 @@ async def test_list_projects_success(mock_mcp, mock_context):
mock_async_client.get.return_value = mock_response
mock_client.return_value.__aenter__.return_value = mock_async_client
result = await list_projects(mock_context)
result = await find_projects(mock_context)
result_data = json.loads(result)
assert result_data["success"] is True
@ -152,10 +154,10 @@ async def test_get_project_not_found(mock_mcp, mock_context):
"""Test getting a non-existent project."""
register_project_tools(mock_mcp)
# Get the get_project function
get_project = mock_mcp._tools.get("get_project")
# Get the find_projects function (used for getting single project)
find_projects = mock_mcp._tools.get("find_projects")
assert get_project is not None, "get_project tool not registered"
assert find_projects is not None, "find_projects tool not registered"
# Mock 404 response
mock_response = MagicMock()
@ -167,7 +169,7 @@ async def test_get_project_not_found(mock_mcp, mock_context):
mock_async_client.get.return_value = mock_response
mock_client.return_value.__aenter__.return_value = mock_async_client
result = await get_project(mock_context, project_id="non-existent")
result = await find_projects(mock_context, project_id="non-existent")
result_data = json.loads(result)
assert result_data["success"] is False

67
python/tests/mcp_server/features/tasks/test_task_tools.py
View File

@ -35,13 +35,13 @@ def mock_context():
@pytest.mark.asyncio
async def test_create_task_with_sources(mock_mcp, mock_context):
"""Test creating a task with sources and code examples."""
"""Test creating a task using manage_task."""
register_task_tools(mock_mcp)
# Get the create_task function
create_task = mock_mcp._tools.get("create_task")
# Get the manage_task function
manage_task = mock_mcp._tools.get("manage_task")
assert create_task is not None, "create_task tool not registered"
assert manage_task is not None, "manage_task tool not registered"
# Mock HTTP response
mock_response = MagicMock()
@ -56,36 +56,35 @@ async def test_create_task_with_sources(mock_mcp, mock_context):
mock_async_client.post.return_value = mock_response
mock_client.return_value.__aenter__.return_value = mock_async_client
result = await create_task(
result = await manage_task(
mock_context,
action="create",
project_id="project-123",
title="Implement OAuth2",
description="Add OAuth2 authentication",
assignee="AI IDE Agent",
sources=[{"url": "https://oauth.net", "type": "doc", "relevance": "OAuth spec"}],
code_examples=[{"file": "auth.py", "function": "authenticate", "purpose": "Example"}],
)
result_data = json.loads(result)
assert result_data["success"] is True
assert result_data["task_id"] == "task-123"
# Verify sources and examples were sent
# Verify the task was created properly
call_args = mock_async_client.post.call_args
sent_data = call_args[1]["json"]
assert len(sent_data["sources"]) == 1
assert len(sent_data["code_examples"]) == 1
assert sent_data["title"] == "Implement OAuth2"
assert sent_data["assignee"] == "AI IDE Agent"
@pytest.mark.asyncio
async def test_list_tasks_with_project_filter(mock_mcp, mock_context):
async def test_find_tasks_with_project_filter(mock_mcp, mock_context):
"""Test listing tasks with project-specific endpoint."""
register_task_tools(mock_mcp)
# Get the list_tasks function
list_tasks = mock_mcp._tools.get("list_tasks")
# Get the find_tasks function
find_tasks = mock_mcp._tools.get("find_tasks")
assert list_tasks is not None, "list_tasks tool not registered"
assert find_tasks is not None, "find_tasks tool not registered"
# Mock HTTP response
mock_response = MagicMock()
@ -102,7 +101,7 @@ async def test_list_tasks_with_project_filter(mock_mcp, mock_context):
mock_async_client.get.return_value = mock_response
mock_client.return_value.__aenter__.return_value = mock_async_client
result = await list_tasks(mock_context, filter_by="project", filter_value="project-123")
result = await find_tasks(mock_context, filter_by="project", filter_value="project-123")
result_data = json.loads(result)
assert result_data["success"] is True
@ -114,11 +113,11 @@ async def test_list_tasks_with_project_filter(mock_mcp, mock_context):
@pytest.mark.asyncio
async def test_list_tasks_with_status_filter(mock_mcp, mock_context):
async def test_find_tasks_with_status_filter(mock_mcp, mock_context):
"""Test listing tasks with status filter uses generic endpoint."""
register_task_tools(mock_mcp)
list_tasks = mock_mcp._tools.get("list_tasks")
find_tasks = mock_mcp._tools.get("find_tasks")
# Mock HTTP response
mock_response = MagicMock()
@ -130,7 +129,7 @@ async def test_list_tasks_with_status_filter(mock_mcp, mock_context):
mock_async_client.get.return_value = mock_response
mock_client.return_value.__aenter__.return_value = mock_async_client
result = await list_tasks(
result = await find_tasks(
mock_context, filter_by="status", filter_value="todo", project_id="project-123"
)
@ -149,10 +148,10 @@ async def test_update_task_status(mock_mcp, mock_context):
"""Test updating task status."""
register_task_tools(mock_mcp)
# Get the update_task function
update_task = mock_mcp._tools.get("update_task")
# Get the manage_task function
manage_task = mock_mcp._tools.get("manage_task")
assert update_task is not None, "update_task tool not registered"
assert manage_task is not None, "manage_task tool not registered"
# Mock HTTP response
mock_response = MagicMock()
@ -167,8 +166,8 @@ async def test_update_task_status(mock_mcp, mock_context):
mock_async_client.put.return_value = mock_response
mock_client.return_value.__aenter__.return_value = mock_async_client
result = await update_task(
mock_context, task_id="task-123", status="doing", assignee="User"
result = await manage_task(
mock_context, action="update", task_id="task-123", status="doing", assignee="User"
)
result_data = json.loads(result)
@ -187,13 +186,13 @@ 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")
# Get the manage_task function
manage_task = mock_mcp._tools.get("manage_task")
assert update_task is not None, "update_task tool not registered"
assert manage_task is not None, "manage_task tool not registered"
# Call update_task with no optional fields
result = await update_task(mock_context, task_id="task-123")
# Call manage_task with update action but no fields to update
result = await manage_task(mock_context, action="update", task_id="task-123")
result_data = json.loads(result)
assert result_data["success"] is False
@ -208,10 +207,10 @@ async def test_delete_task_already_archived(mock_mcp, mock_context):
"""Test deleting an already archived task."""
register_task_tools(mock_mcp)
# Get the delete_task function
delete_task = mock_mcp._tools.get("delete_task")
# Get the manage_task function
manage_task = mock_mcp._tools.get("manage_task")
assert delete_task is not None, "delete_task tool not registered"
assert manage_task is not None, "manage_task tool not registered"
# Mock 400 response for already archived
mock_response = MagicMock()
@ -223,7 +222,7 @@ async def test_delete_task_already_archived(mock_mcp, mock_context):
mock_async_client.delete.return_value = mock_response
mock_client.return_value.__aenter__.return_value = mock_async_client
result = await delete_task(mock_context, task_id="task-123")
result = await manage_task(mock_context, action="delete", task_id="task-123")
result_data = json.loads(result)
assert result_data["success"] is False
@ -232,5 +231,5 @@ async def test_delete_task_already_archived(mock_mcp, mock_context):
assert isinstance(result_data["error"], dict), (
"Error should be structured format, not string"
)
assert result_data["error"]["type"] == "already_archived"
assert "already archived" in result_data["error"]["message"].lower()
assert result_data["error"]["type"] == "http_error"
assert "http 400" in result_data["error"]["message"].lower()