from typing import Any
from dooray_mcp.services.client import dooray_client
from dooray_mcp.types.task import (
CreateTaskParams,
ListTasksParams,
TaskDetail,
TaskListResult,
UpdateTaskParams,
)
async def list_tasks(project_id: str, params: ListTasksParams) -> TaskListResult:
path = f"/project/v1/projects/{project_id}/posts"
query_params: dict[str, Any] = {
"page": params.page,
"size": params.size,
"order": params.order,
}
if params.workflow_class:
query_params["postWorkflowClasses"] = params.workflow_class
if params.to_member_ids:
query_params["toMemberIds"] = params.to_member_ids
if params.tag_ids:
query_params["tagIds"] = params.tag_ids
if params.milestone_ids:
query_params["milestoneIds"] = params.milestone_ids
if params.due_date:
query_params["dueAt"] = params.due_date
response = await dooray_client.get(path, params=query_params)
result = response.get("result", [])
if isinstance(result, list):
total_count = response.get("header", {}).get("totalCount", len(result))
return TaskListResult(totalCount=total_count, contents=result)
return TaskListResult.model_validate(result)
async def get_task(project_id: str, post_id: str) -> TaskDetail:
path = f"/project/v1/projects/{project_id}/posts/{post_id}"
response = await dooray_client.get(path)
return TaskDetail.model_validate(response.get("result", {}))
async def create_task(project_id: str, params: CreateTaskParams) -> TaskDetail:
path = f"/project/v1/projects/{project_id}/posts"
body: dict[str, Any] = {
"subject": params.subject,
}
if params.body:
body["body"] = {
"mimeType": params.body_mime_type,
"content": params.body,
}
if params.to_member_ids:
body["users"] = {
"to": [
{"type": "member", "member": {"organizationMemberId": mid}}
for mid in params.to_member_ids
]
}
if params.cc_member_ids:
body["users"] = body.get("users", {})
body["users"]["cc"] = [
{"type": "member", "member": {"organizationMemberId": mid}}
for mid in params.cc_member_ids
]
if params.milestone_id:
body["milestoneId"] = params.milestone_id
if params.tag_ids:
body["tagIds"] = params.tag_ids
if params.due_date:
body["dueDate"] = params.due_date
if params.priority:
body["priority"] = params.priority
if params.parent_post_id:
body["parentPostId"] = params.parent_post_id
response = await dooray_client.post(path, json_data=body)
return TaskDetail.model_validate(response.get("result", {}))
async def update_task(project_id: str, post_id: str, params: UpdateTaskParams) -> TaskDetail:
path = f"/project/v1/projects/{project_id}/posts/{post_id}"
body: dict[str, Any] = {}
if params.subject is not None:
body["subject"] = params.subject
if params.body is not None:
body["body"] = {
"mimeType": params.body_mime_type,
"content": params.body,
}
if params.due_date is not None:
body["dueDate"] = params.due_date
if params.priority is not None:
body["priority"] = params.priority
if params.milestone_id is not None:
body["milestoneId"] = params.milestone_id
response = await dooray_client.put(path, json_data=body)
return TaskDetail.model_validate(response.get("result", {}))
async def set_task_workflow(project_id: str, post_id: str, workflow_id: str) -> dict[str, Any]:
path = f"/project/v1/projects/{project_id}/posts/{post_id}/set-workflow"
body = {"workflowId": workflow_id}
return await dooray_client.post(path, json_data=body)
async def set_task_done(project_id: str, post_id: str) -> dict[str, Any]:
path = f"/project/v1/projects/{project_id}/posts/{post_id}/set-done"
return await dooray_client.post(path, json_data={})
