Add document and version management tools
Extract document management functionality into focused tools: - create_document: Create new documents with metadata - list_documents: List all documents in a project - get_document: Retrieve specific document details - update_document: Modify existing documents - delete_document: Remove documents from projects Extract version control functionality: - create_version: Create immutable snapshots - list_versions: View version history - get_version: Retrieve specific version content - restore_version: Rollback to previous versions Includes improved documentation and error messages based on testing.
This commit is contained in:
Rasmus Widing
parent
b2cab81346
commit
4f317d9ff5
12
python/src/mcp_server/features/documents/__init__.py
Normal file
12
python/src/mcp_server/features/documents/__init__.py
Normal file
@ -0,0 +1,12 @@
|
||||
"""
|
||||
Document and version management tools for Archon MCP Server.
|
||||
|
||||
This module provides separate tools for document operations:
|
||||
- create_document, list_documents, get_document, update_document, delete_document
|
||||
- create_version, list_versions, get_version, restore_version
|
||||
"""
|
||||
|
||||
from .document_tools import register_document_tools
|
||||
from .version_tools import register_version_tools
|
||||
|
||||
__all__ = ["register_document_tools", "register_version_tools"]
|
||||
298
python/src/mcp_server/features/documents/document_tools.py
Normal file
298
python/src/mcp_server/features/documents/document_tools.py
Normal file
@ -0,0 +1,298 @@
|
||||
"""
|
||||
Simple 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.
|
||||
"""
|
||||
|
||||
import json
|
||||
import logging
|
||||
from typing import Any, Optional, Dict, List
|
||||
from urllib.parse import urljoin
|
||||
|
||||
import httpx
|
||||
from mcp.server.fastmcp import Context, FastMCP
|
||||
|
||||
from src.server.config.service_discovery import get_api_url
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def register_document_tools(mcp: FastMCP):
|
||||
"""Register individual document management tools with the MCP server."""
|
||||
|
||||
@mcp.tool()
|
||||
async def create_document(
|
||||
ctx: Context,
|
||||
project_id: str,
|
||||
title: str,
|
||||
document_type: str,
|
||||
content: Optional[Dict[str, Any]] = None,
|
||||
tags: Optional[List[str]] = None,
|
||||
author: Optional[str] = None,
|
||||
) -> str:
|
||||
"""
|
||||
Create a new document with automatic versioning.
|
||||
|
||||
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)
|
||||
|
||||
Returns:
|
||||
JSON with document details:
|
||||
{
|
||||
"success": true,
|
||||
"document": {...},
|
||||
"document_id": "doc-123",
|
||||
"message": "Document created successfully"
|
||||
}
|
||||
|
||||
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"}
|
||||
}
|
||||
)
|
||||
"""
|
||||
try:
|
||||
api_url = get_api_url()
|
||||
timeout = httpx.Timeout(30.0, connect=5.0)
|
||||
|
||||
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:
|
||||
error_detail = response.text
|
||||
return json.dumps({"success": False, "error": error_detail})
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error creating document: {e}")
|
||||
return json.dumps({"success": False, "error": str(e)})
|
||||
|
||||
@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 = httpx.Timeout(30.0, connect=5.0)
|
||||
|
||||
async with httpx.AsyncClient(timeout=timeout) as client:
|
||||
response = await client.get(urljoin(api_url, f"/api/projects/{project_id}/docs"))
|
||||
|
||||
if response.status_code == 200:
|
||||
result = response.json()
|
||||
return json.dumps({
|
||||
"success": True,
|
||||
"documents": result.get("documents", []),
|
||||
"count": len(result.get("documents", [])),
|
||||
})
|
||||
else:
|
||||
return json.dumps({
|
||||
"success": False,
|
||||
"error": f"HTTP {response.status_code}: {response.text}",
|
||||
})
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error listing documents: {e}")
|
||||
return json.dumps({"success": False, "error": str(e)})
|
||||
|
||||
@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 = httpx.Timeout(30.0, connect=5.0)
|
||||
|
||||
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 json.dumps({"success": False, "error": f"Document {doc_id} not found"})
|
||||
else:
|
||||
return json.dumps({"success": False, "error": "Failed to get document"})
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting document: {e}")
|
||||
return json.dumps({"success": False, "error": str(e)})
|
||||
|
||||
@mcp.tool()
|
||||
async def update_document(
|
||||
ctx: Context,
|
||||
project_id: str,
|
||||
doc_id: str,
|
||||
title: Optional[str] = None,
|
||||
content: Optional[Dict[str, Any]] = None,
|
||||
tags: Optional[List[str]] = None,
|
||||
author: Optional[str] = None,
|
||||
) -> str:
|
||||
"""
|
||||
Update a document's properties.
|
||||
|
||||
Args:
|
||||
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"})
|
||||
"""
|
||||
try:
|
||||
api_url = get_api_url()
|
||||
timeout = httpx.Timeout(30.0, connect=5.0)
|
||||
|
||||
# Build update fields
|
||||
update_fields = {}
|
||||
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:
|
||||
error_detail = response.text
|
||||
return json.dumps({"success": False, "error": error_detail})
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error updating document: {e}")
|
||||
return json.dumps({"success": False, "error": str(e)})
|
||||
|
||||
@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 = httpx.Timeout(30.0, connect=5.0)
|
||||
|
||||
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 json.dumps({"success": False, "error": f"Document {doc_id} not found"})
|
||||
else:
|
||||
return json.dumps({"success": False, "error": "Failed to delete document"})
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error deleting document: {e}")
|
||||
return json.dumps({"success": False, "error": str(e)})
|
||||
307
python/src/mcp_server/features/documents/version_tools.py
Normal file
307
python/src/mcp_server/features/documents/version_tools.py
Normal file
@ -0,0 +1,307 @@
|
||||
"""
|
||||
Simple version management tools for Archon MCP Server.
|
||||
|
||||
Provides separate, focused tools for version control operations.
|
||||
Supports versioning of documents, features, and other project data.
|
||||
"""
|
||||
|
||||
import json
|
||||
import logging
|
||||
from typing import Any, Optional
|
||||
from urllib.parse import urljoin
|
||||
|
||||
import httpx
|
||||
from mcp.server.fastmcp import Context, FastMCP
|
||||
|
||||
from src.server.config.service_discovery import get_api_url
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def register_version_tools(mcp: FastMCP):
|
||||
"""Register individual version management tools with the MCP server."""
|
||||
|
||||
@mcp.tool()
|
||||
async def create_version(
|
||||
ctx: Context,
|
||||
project_id: str,
|
||||
field_name: str,
|
||||
content: Any,
|
||||
change_summary: Optional[str] = None,
|
||||
document_id: Optional[str] = None,
|
||||
created_by: str = "system",
|
||||
) -> 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 = httpx.Timeout(30.0, connect=5.0)
|
||||
|
||||
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 json.dumps({
|
||||
"success": False,
|
||||
"error": f"Invalid field_name '{field_name}'. Must be one of: docs, features, data, or prd"
|
||||
})
|
||||
elif "content" in error_text and "required" in error_text:
|
||||
return json.dumps({
|
||||
"success": False,
|
||||
"error": "Content is required and cannot be empty. Provide the complete data to version."
|
||||
})
|
||||
elif "format" in error_text or "type" in error_text:
|
||||
if field_name == "docs":
|
||||
return json.dumps({
|
||||
"success": False,
|
||||
"error": f"For field_name='docs', content must be an array. Example: [{{'id': 'doc1', 'title': 'Guide', 'content': {{...}}}}]"
|
||||
})
|
||||
else:
|
||||
return json.dumps({
|
||||
"success": False,
|
||||
"error": f"For field_name='{field_name}', content must be a dictionary/object. Example: {{'key': 'value'}}"
|
||||
})
|
||||
return json.dumps({"success": False, "error": f"Bad request: {response.text}"})
|
||||
elif response.status_code == 404:
|
||||
return json.dumps({
|
||||
"success": False,
|
||||
"error": f"Project {project_id} not found. Please check the project ID."
|
||||
})
|
||||
else:
|
||||
return json.dumps({
|
||||
"success": False,
|
||||
"error": f"Failed to create version (HTTP {response.status_code}): {response.text}"
|
||||
})
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error creating version: {e}")
|
||||
return json.dumps({"success": False, "error": str(e)})
|
||||
|
||||
@mcp.tool()
|
||||
async def list_versions(
|
||||
ctx: Context,
|
||||
project_id: str,
|
||||
field_name: Optional[str] = None
|
||||
) -> str:
|
||||
"""
|
||||
List version history for a project.
|
||||
|
||||
Args:
|
||||
project_id: Project UUID (required)
|
||||
field_name: Filter by field name - "docs", "features", "data", "prd" (optional)
|
||||
|
||||
Returns:
|
||||
JSON array of versions with metadata
|
||||
|
||||
Example:
|
||||
list_versions(project_id="uuid", field_name="docs")
|
||||
"""
|
||||
try:
|
||||
api_url = get_api_url()
|
||||
timeout = httpx.Timeout(30.0, connect=5.0)
|
||||
|
||||
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
|
||||
)
|
||||
|
||||
if response.status_code == 200:
|
||||
result = response.json()
|
||||
return json.dumps({
|
||||
"success": True,
|
||||
"versions": result.get("versions", []),
|
||||
"count": len(result.get("versions", []))
|
||||
})
|
||||
else:
|
||||
return json.dumps({
|
||||
"success": False,
|
||||
"error": f"HTTP {response.status_code}: {response.text}"
|
||||
})
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error listing versions: {e}")
|
||||
return json.dumps({"success": False, "error": str(e)})
|
||||
|
||||
@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 = httpx.Timeout(30.0, connect=5.0)
|
||||
|
||||
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 json.dumps({
|
||||
"success": False,
|
||||
"error": f"Version {version_number} not found for field {field_name}"
|
||||
})
|
||||
else:
|
||||
return json.dumps({
|
||||
"success": False,
|
||||
"error": "Failed to get version"
|
||||
})
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting version: {e}")
|
||||
return json.dumps({"success": False, "error": str(e)})
|
||||
|
||||
@mcp.tool()
|
||||
async def restore_version(
|
||||
ctx: Context,
|
||||
project_id: str,
|
||||
field_name: str,
|
||||
version_number: int,
|
||||
restored_by: str = "system",
|
||||
) -> str:
|
||||
"""
|
||||
Restore a previous version.
|
||||
|
||||
Args:
|
||||
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)
|
||||
"""
|
||||
try:
|
||||
api_url = get_api_url()
|
||||
timeout = httpx.Timeout(30.0, connect=5.0)
|
||||
|
||||
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 json.dumps({
|
||||
"success": False,
|
||||
"error": f"Version {version_number} not found for field {field_name}"
|
||||
})
|
||||
else:
|
||||
error_detail = response.text
|
||||
return json.dumps({"success": False, "error": error_detail})
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error restoring version: {e}")
|
||||
return json.dumps({"success": False, "error": str(e)})
|
||||
Loading…
Reference in New Issue
Block a user