Tasks MCP Server

from mcp.server.fastmcp import Context, FastMCP from mcp.server.session import ServerSession from db.tasks_interfaces import DatabaseABC from models.server_context import AppContext from models.tasks import Task, TaskStatus def create_tasks_tools(mcp: FastMCP) -> None: @mcp.tool() async def tasks_list_tool( ctx: Context[ServerSession, AppContext], task_limit: int = 10 ) -> list[Task]: """Get a list of tasks. Args: task_limit (int): The maximum number of tasks to return. Default is 10. Returns: list[Task]: A list of tasks. """ database: DatabaseABC = ctx.request_context.lifespan_context.db tasks = database.get_tasks(limit=task_limit) return [Task(**task.__dict__) for task in tasks] @mcp.tool() async def get_task_tool( ctx: Context[ServerSession, AppContext], task_id: int ) -> Task | None: """Get a task by ID. Args: task_id (int): The ID of the task to retrieve. Returns: Task: The task object if found, otherwise None. """ database: DatabaseABC = ctx.request_context.lifespan_context.db task = database.get_task(task_id=task_id) return Task(**task.__dict__) if task else None @mcp.tool() async def add_task_tool( ctx: Context[ServerSession, AppContext], title: str, description: str | None = None, ) -> Task: """Add a new task. Args: title (str): The title of the task. description (str): The description of the task. optional value. Returns: Task: The newly created task object. """ database: DatabaseABC = ctx.request_context.lifespan_context.db # Normalize None to empty string for storage compatibility task = database.add_task(title=title, description=description or "") return Task(**task.__dict__) @mcp.tool() async def update_task_tool( ctx: Context[ServerSession, AppContext], id: int, title: str | None = None, description: str | None = None, status: TaskStatus | None = None, ) -> Task: """Update an existing task. Args: task (Task): The task object containing updated information. Returns: Task: The updated task object. """ database: DatabaseABC = ctx.request_context.lifespan_context.db task = database.get_task(task_id=id) if title is None or title == "": title = task.title if description is None or description == "": description = task.description if status is None: status = task.status task = database.update_task( task_id=id, title=title, description=description, status=status, ) return Task(**task.__dict__) @mcp.tool() async def delete_task_tool( ctx: Context[ServerSession, AppContext], task_id: int ) -> str: """Delete a task by ID. Args: task_id (int): The ID of the task to delete. Returns: None """ database: DatabaseABC = ctx.request_context.lifespan_context.db database.delete_task(task_id=task_id) return f"Task {task_id} deleted successfully." @mcp.tool() async def filter_tasks_tool( ctx: Context[ServerSession, AppContext], status: TaskStatus | None = None, title: str | None = None, description: str | None = None, created_at: str | None = None, ) -> list[Task]: """Filter tasks by status and title. Args: status (TaskStatus): The status to filter tasks. Optional value. title (str): The title to filter tasks. Optional value. description (str): The description to filter tasks. Optional value. created_at (str): The creation date to filter tasks. Optional value. Returns: list[Task]: A list of tasks matching the specified status and title. """ database: DatabaseABC = ctx.request_context.lifespan_context.db tasks = database.filter_tasks( title=title or "", description=description or "", status=status if status is not None else None, created_at=created_at or "", ) return [Task(**task.__dict__) for task in tasks] @mcp.tool() async def get_tasks_status() -> dict: """Get the status tasks values. Returns: dict: The status tasks values. """ return {status.name: status.value for status in TaskStatus}