import asyncio
import typer
from rich.console import Console
from dooray_mcp.controllers import (
comment as comment_controller,
)
from dooray_mcp.controllers import (
member as member_controller,
)
from dooray_mcp.controllers import (
messenger as messenger_controller,
)
from dooray_mcp.controllers import (
milestone as milestone_controller,
)
from dooray_mcp.controllers import (
project as project_controller,
)
from dooray_mcp.controllers import (
tag as tag_controller,
)
from dooray_mcp.controllers import (
task as task_controller,
)
from dooray_mcp.controllers import (
template as template_controller,
)
from dooray_mcp.utils.config import config
from dooray_mcp.utils.errors import DoorayError
app = typer.Typer(name="dooray", help="Dooray CLI")
task_app = typer.Typer(name="task", help="Task commands")
comment_app = typer.Typer(name="comment", help="Comment commands")
project_app = typer.Typer(name="project", help="Project commands")
milestone_app = typer.Typer(name="milestone", help="Milestone commands")
tag_app = typer.Typer(name="tag", help="Tag commands")
template_app = typer.Typer(name="template", help="Template commands")
member_app = typer.Typer(name="member", help="Member commands")
messenger_app = typer.Typer(name="messenger", help="Messenger commands")
app.add_typer(task_app)
app.add_typer(comment_app)
app.add_typer(project_app)
app.add_typer(milestone_app)
app.add_typer(tag_app)
app.add_typer(template_app)
app.add_typer(member_app)
app.add_typer(messenger_app)
console = Console()
def run_async(coro):
return asyncio.get_event_loop().run_until_complete(coro)
def handle_error(e: Exception):
if isinstance(e, DoorayError):
console.print(f"[red]Error: {e}[/red]")
else:
console.print(f"[red]Unexpected error: {e}[/red]")
raise typer.Exit(1)
@task_app.command("list")
def list_tasks(
project_id: str = typer.Option(..., "--project-id", "-p"),
page: int = typer.Option(0, "--page"),
size: int = typer.Option(20, "--size"),
workflow: str | None = typer.Option(None, "--workflow", "-w"),
order: str = typer.Option("-createdAt", "--order"),
):
config.load()
try:
result = run_async(
task_controller.list_tasks(
project_id=project_id,
page=page,
size=size,
workflow_class=workflow,
order=order,
)
)
console.print(result.content)
except Exception as e:
handle_error(e)
@task_app.command("get")
def get_task(
post_id: str = typer.Argument(..., help="Task ID"),
project_id: str = typer.Option(..., "--project-id", "-p"),
):
config.load()
try:
result = run_async(task_controller.get_task(project_id=project_id, post_id=post_id))
console.print(result.content)
except Exception as e:
handle_error(e)
@task_app.command("create")
def create_task(
project_id: str = typer.Option(..., "--project-id", "-p"),
subject: str = typer.Option(..., "--subject", "-s"),
body: str | None = typer.Option(None, "--body", "-b"),
to_member_ids: str | None = typer.Option(None, "--to", help="Comma-separated member IDs"),
priority: str | None = typer.Option(None, "--priority"),
due_date: str | None = typer.Option(None, "--due"),
milestone_id: str | None = typer.Option(None, "--milestone"),
parent_post_id: str | None = typer.Option(None, "--parent"),
):
config.load()
try:
to_ids = to_member_ids.split(",") if to_member_ids else None
result = run_async(
task_controller.create_task(
project_id=project_id,
subject=subject,
body=body,
to_member_ids=to_ids,
priority=priority,
due_date=due_date,
milestone_id=milestone_id,
parent_post_id=parent_post_id,
)
)
console.print(f"[green]{result.content}[/green]")
except Exception as e:
handle_error(e)
@task_app.command("update")
def update_task(
post_id: str = typer.Argument(..., help="Task ID"),
project_id: str = typer.Option(..., "--project-id", "-p"),
subject: str | None = typer.Option(None, "--subject", "-s"),
body: str | None = typer.Option(None, "--body", "-b"),
due_date: str | None = typer.Option(None, "--due"),
priority: str | None = typer.Option(None, "--priority"),
milestone_id: str | None = typer.Option(None, "--milestone"),
):
config.load()
try:
result = run_async(
task_controller.update_task(
project_id=project_id,
post_id=post_id,
subject=subject,
body=body,
due_date=due_date,
priority=priority,
milestone_id=milestone_id,
)
)
console.print(f"[green]{result.content}[/green]")
except Exception as e:
handle_error(e)
@task_app.command("set-workflow")
def set_task_workflow(
post_id: str = typer.Argument(..., help="Task ID"),
workflow_id: str = typer.Option(..., "--workflow-id", "-w"),
project_id: str = typer.Option(..., "--project-id", "-p"),
):
config.load()
try:
result = run_async(
task_controller.set_task_workflow(
project_id=project_id,
post_id=post_id,
workflow_id=workflow_id,
)
)
console.print(f"[green]{result.content}[/green]")
except Exception as e:
handle_error(e)
@task_app.command("set-done")
def set_task_done(
post_id: str = typer.Argument(..., help="Task ID"),
project_id: str = typer.Option(..., "--project-id", "-p"),
):
config.load()
try:
result = run_async(task_controller.set_task_done(project_id=project_id, post_id=post_id))
console.print(f"[green]{result.content}[/green]")
except Exception as e:
handle_error(e)
@comment_app.command("list")
def list_comments(
post_id: str = typer.Argument(..., help="Task ID"),
project_id: str = typer.Option(..., "--project-id", "-p"),
):
config.load()
try:
result = run_async(comment_controller.list_comments(project_id=project_id, post_id=post_id))
console.print(result.content)
except Exception as e:
handle_error(e)
@comment_app.command("add")
def add_comment(
post_id: str = typer.Argument(..., help="Task ID"),
content: str = typer.Option(..., "--content", "-c", help="Comment content"),
project_id: str = typer.Option(..., "--project-id", "-p"),
):
config.load()
try:
result = run_async(
comment_controller.add_comment(project_id=project_id, post_id=post_id, content=content)
)
console.print(f"[green]{result.content}[/green]")
except Exception as e:
handle_error(e)
@comment_app.command("update")
def update_comment(
post_id: str = typer.Argument(..., help="Task ID"),
log_id: str = typer.Argument(..., help="Comment ID"),
content: str = typer.Option(..., "--content", "-c", help="New content"),
project_id: str = typer.Option(..., "--project-id", "-p"),
):
config.load()
try:
result = run_async(
comment_controller.update_comment(
project_id=project_id, post_id=post_id, log_id=log_id, content=content
)
)
console.print(f"[green]{result.content}[/green]")
except Exception as e:
handle_error(e)
@comment_app.command("delete")
def delete_comment(
post_id: str = typer.Argument(..., help="Task ID"),
log_id: str = typer.Argument(..., help="Comment ID"),
project_id: str = typer.Option(..., "--project-id", "-p"),
):
config.load()
try:
result = run_async(
comment_controller.delete_comment(project_id=project_id, post_id=post_id, log_id=log_id)
)
console.print(f"[green]{result.content}[/green]")
except Exception as e:
handle_error(e)
@project_app.command("get")
def get_project(
project_id: str = typer.Argument(..., help="Project ID"),
):
config.load()
try:
result = run_async(project_controller.get_project(project_id=project_id))
console.print(result.content)
except Exception as e:
handle_error(e)
@project_app.command("workflows")
def list_workflows(
project_id: str = typer.Argument(..., help="Project ID"),
):
config.load()
try:
result = run_async(project_controller.list_workflows(project_id=project_id))
console.print(result.content)
except Exception as e:
handle_error(e)
@milestone_app.command("list")
def list_milestones(
project_id: str = typer.Option(..., "--project-id", "-p"),
page: int = typer.Option(0, "--page"),
size: int = typer.Option(20, "--size"),
status: str | None = typer.Option(None, "--status"),
):
config.load()
try:
result = run_async(
milestone_controller.list_milestones(
project_id=project_id,
page=page,
size=size,
status=status,
)
)
console.print(result.content)
except Exception as e:
handle_error(e)
@milestone_app.command("get")
def get_milestone(
milestone_id: str = typer.Argument(..., help="Milestone ID"),
project_id: str = typer.Option(..., "--project-id", "-p"),
):
config.load()
try:
result = run_async(
milestone_controller.get_milestone(
project_id=project_id,
milestone_id=milestone_id,
)
)
console.print(result.content)
except Exception as e:
handle_error(e)
@milestone_app.command("create")
def create_milestone(
project_id: str = typer.Option(..., "--project-id", "-p"),
name: str = typer.Option(..., "--name", "-n"),
start_at: str | None = typer.Option(None, "--start"),
end_at: str | None = typer.Option(None, "--end"),
):
config.load()
try:
result = run_async(
milestone_controller.create_milestone(
project_id=project_id,
name=name,
start_at=start_at,
end_at=end_at,
)
)
console.print(f"[green]{result.content}[/green]")
except Exception as e:
handle_error(e)
@milestone_app.command("update")
def update_milestone(
milestone_id: str = typer.Argument(..., help="Milestone ID"),
project_id: str = typer.Option(..., "--project-id", "-p"),
name: str | None = typer.Option(None, "--name", "-n"),
status: str | None = typer.Option(None, "--status"),
start_at: str | None = typer.Option(None, "--start"),
end_at: str | None = typer.Option(None, "--end"),
):
config.load()
try:
result = run_async(
milestone_controller.update_milestone(
project_id=project_id,
milestone_id=milestone_id,
name=name,
status=status,
start_at=start_at,
end_at=end_at,
)
)
console.print(f"[green]{result.content}[/green]")
except Exception as e:
handle_error(e)
@milestone_app.command("delete")
def delete_milestone(
milestone_id: str = typer.Argument(..., help="Milestone ID"),
project_id: str = typer.Option(..., "--project-id", "-p"),
):
config.load()
try:
result = run_async(
milestone_controller.delete_milestone(
project_id=project_id,
milestone_id=milestone_id,
)
)
console.print(f"[green]{result.content}[/green]")
except Exception as e:
handle_error(e)
@tag_app.command("list")
def list_tags(
project_id: str = typer.Option(..., "--project-id", "-p"),
page: int = typer.Option(0, "--page"),
size: int = typer.Option(20, "--size"),
):
config.load()
try:
result = run_async(
tag_controller.list_tags(
project_id=project_id,
page=page,
size=size,
)
)
console.print(result.content)
except Exception as e:
handle_error(e)
@tag_app.command("get")
def get_tag(
tag_id: str = typer.Argument(..., help="Tag ID"),
project_id: str = typer.Option(..., "--project-id", "-p"),
):
config.load()
try:
result = run_async(
tag_controller.get_tag(
project_id=project_id,
tag_id=tag_id,
)
)
console.print(result.content)
except Exception as e:
handle_error(e)
@tag_app.command("create")
def create_tag(
project_id: str = typer.Option(..., "--project-id", "-p"),
name: str = typer.Option(..., "--name", "-n"),
color: str | None = typer.Option(None, "--color"),
):
config.load()
try:
result = run_async(
tag_controller.create_tag(
project_id=project_id,
name=name,
color=color,
)
)
console.print(f"[green]{result.content}[/green]")
except Exception as e:
handle_error(e)
@template_app.command("list")
def list_templates(
project_id: str = typer.Option(..., "--project-id", "-p"),
page: int = typer.Option(0, "--page"),
size: int = typer.Option(20, "--size"),
):
config.load()
try:
result = run_async(
template_controller.list_templates(
project_id=project_id,
page=page,
size=size,
)
)
console.print(result.content)
except Exception as e:
handle_error(e)
@template_app.command("get")
def get_template(
template_id: str = typer.Argument(..., help="Template ID"),
project_id: str = typer.Option(..., "--project-id", "-p"),
):
config.load()
try:
result = run_async(
template_controller.get_template(
project_id=project_id,
template_id=template_id,
)
)
console.print(result.content)
except Exception as e:
handle_error(e)
@template_app.command("create")
def create_template(
project_id: str = typer.Option(..., "--project-id", "-p"),
name: str = typer.Option(..., "--name", "-n"),
body: str = typer.Option(..., "--body", "-b"),
body_mime_type: str = typer.Option("text/x-markdown", "--body-mime-type"),
):
config.load()
try:
result = run_async(
template_controller.create_template(
project_id=project_id,
name=name,
body=body,
body_mime_type=body_mime_type,
)
)
console.print(f"[green]{result.content}[/green]")
except Exception as e:
handle_error(e)
@template_app.command("update")
def update_template(
template_id: str = typer.Argument(..., help="Template ID"),
project_id: str = typer.Option(..., "--project-id", "-p"),
name: str | None = typer.Option(None, "--name", "-n"),
body: str | None = typer.Option(None, "--body", "-b"),
body_mime_type: str = typer.Option("text/x-markdown", "--body-mime-type"),
):
config.load()
try:
result = run_async(
template_controller.update_template(
project_id=project_id,
template_id=template_id,
name=name,
body=body,
body_mime_type=body_mime_type,
)
)
console.print(f"[green]{result.content}[/green]")
except Exception as e:
handle_error(e)
@template_app.command("delete")
def delete_template(
template_id: str = typer.Argument(..., help="Template ID"),
project_id: str = typer.Option(..., "--project-id", "-p"),
):
config.load()
try:
result = run_async(
template_controller.delete_template(
project_id=project_id,
template_id=template_id,
)
)
console.print(f"[green]{result.content}[/green]")
except Exception as e:
handle_error(e)
@member_app.command("search")
def search_members(
name: str | None = typer.Option(None, "--name"),
user_code: str | None = typer.Option(None, "--user-code"),
external_email: str | None = typer.Option(None, "--external-email"),
page: int = typer.Option(0, "--page"),
size: int = typer.Option(20, "--size"),
):
config.load()
try:
result = run_async(
member_controller.search_members(
name=name,
user_code=user_code,
external_email=external_email,
page=page,
size=size,
)
)
console.print(result.content)
except Exception as e:
handle_error(e)
@member_app.command("list")
def list_project_members(
project_id: str = typer.Option(..., "--project-id", "-p"),
page: int = typer.Option(0, "--page"),
size: int = typer.Option(20, "--size"),
):
config.load()
try:
result = run_async(
member_controller.list_project_members(
project_id=project_id,
page=page,
size=size,
)
)
console.print(result.content)
except Exception as e:
handle_error(e)
@member_app.command("get")
def get_project_member(
member_id: str = typer.Argument(..., help="Member ID"),
project_id: str = typer.Option(..., "--project-id", "-p"),
):
config.load()
try:
result = run_async(
member_controller.get_project_member(
project_id=project_id,
member_id=member_id,
)
)
console.print(result.content)
except Exception as e:
handle_error(e)
@member_app.command("add")
def add_project_member(
project_id: str = typer.Option(..., "--project-id", "-p"),
member_id: str = typer.Option(..., "--member-id", "-m"),
role: str = typer.Option("member", "--role"),
):
config.load()
try:
result = run_async(
member_controller.add_project_member(
project_id=project_id,
member_id=member_id,
role=role,
)
)
console.print(f"[green]{result.content}[/green]")
except Exception as e:
handle_error(e)
@messenger_app.command("channels")
def list_channels():
config.load()
try:
result = run_async(messenger_controller.list_channels())
console.print(result.content)
except Exception as e:
handle_error(e)
@messenger_app.command("create-channel")
def create_channel(
title: str = typer.Option(..., "--title", "-t"),
member_ids: str = typer.Option(..., "--members", help="Comma-separated member IDs"),
channel_type: str = typer.Option("private", "--type"),
capacity: int | None = typer.Option(None, "--capacity"),
):
config.load()
try:
split_member_ids = member_ids.split(",")
result = run_async(
messenger_controller.create_channel(
title=title,
member_ids=split_member_ids,
channel_type=channel_type,
capacity=capacity,
)
)
console.print(f"[green]{result.content}[/green]")
except Exception as e:
handle_error(e)
@messenger_app.command("send")
def send_channel_message(
channel_id: str = typer.Argument(..., help="Channel ID"),
content: str = typer.Option(..., "--content", "-c", help="Message content"),
):
config.load()
try:
result = run_async(
messenger_controller.send_channel_message(
channel_id=channel_id,
content=content,
)
)
console.print(f"[green]{result.content}[/green]")
except Exception as e:
handle_error(e)
@messenger_app.command("dm")
def send_direct_message(
member_id: str = typer.Argument(..., help="Member ID"),
content: str = typer.Option(..., "--content", "-c", help="Message content"),
):
config.load()
try:
result = run_async(
messenger_controller.send_direct_message(
member_id=member_id,
content=content,
)
)
console.print(f"[green]{result.content}[/green]")
except Exception as e:
handle_error(e)
@messenger_app.command("join")
def join_channel_members(
channel_id: str = typer.Argument(..., help="Channel ID"),
member_ids: str = typer.Option(..., "--members", help="Comma-separated member IDs"),
):
config.load()
try:
split_member_ids = member_ids.split(",")
result = run_async(
messenger_controller.join_channel_members(
channel_id=channel_id,
member_ids=split_member_ids,
)
)
console.print(f"[green]{result.content}[/green]")
except Exception as e:
handle_error(e)
@messenger_app.command("leave")
def leave_channel_members(
channel_id: str = typer.Argument(..., help="Channel ID"),
member_ids: str = typer.Option(..., "--members", help="Comma-separated member IDs"),
):
config.load()
try:
split_member_ids = member_ids.split(",")
result = run_async(
messenger_controller.leave_channel_members(
channel_id=channel_id,
member_ids=split_member_ids,
)
)
console.print(f"[green]{result.content}[/green]")
except Exception as e:
handle_error(e)
if __name__ == "__main__":
app()
