tasks.py•4.41 kB
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:
"""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 = ""
) -> Task:
"""Add a new task.
Args:
title (str): The title of the task.
description (str): The description of the task. Default is an empty string.
Returns:
Task: The newly created task object.
"""
database: DatabaseABC = ctx.request_context.lifespan_context.db
task = database.add_task(title=title, description=description)
return Task(**task.__dict__)
@mcp.tool()
async def update_task_tool(
ctx: Context[ServerSession, AppContext],
id: int,
title: str,
description: str,
status: TaskStatus,
) -> 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,
title: str,
description: str,
created_at: str,
) -> list[Task]:
"""Filter tasks by status and title.
Args:
status (TaskStatus): The status to filter tasks by.
title (str): The title to filter tasks by.
description (str): The description to filter tasks by.
created_at (str): The creation date to filter tasks by.
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, description=description, status=status, created_at=created_at
)
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}