server.py•6.16 kB
#!/usr/bin/env python3
from typing import Any
from mcp.server.lowlevel import NotificationOptions, Server
from mcp.server.models import InitializationOptions
import mcp.server.stdio
import mcp.types as types
from .clickup_client import ClickUpClient
from .tools.tasks import TaskTools
from .tools.workspaces import WorkspaceTools
from .tools.lists import ListTools
class ClickUpMCPServer:
"""ClickUp MCP Server"""
def __init__(self):
self.server = Server("clickup-mcp")
self.client = None
self.task_tools = None
self.workspace_tools = None
self.list_tools = None
self._setup_handlers()
def _setup_handlers(self):
"""Setup MCP server handlers"""
@self.server.list_resources()
async def handle_list_resources() -> list[types.Resource]:
"""List available resources"""
return [
types.Resource(
uri="clickup://workspaces",
name="Workspaces",
description="List of all accessible workspaces",
mimeType="application/json",
),
types.Resource(
uri="clickup://help",
name="ClickUp MCP Help",
description="Help and usage information for ClickUp MCP server",
mimeType="text/plain",
),
]
@self.server.read_resource()
async def handle_read_resource(uri: types.AnyUrl) -> str:
"""Read a specific resource"""
if uri.scheme != "clickup":
raise ValueError(f"Unsupported URI scheme: {uri.scheme}")
path = str(uri).replace("clickup://", "")
if path == "workspaces":
if not self.client:
raise RuntimeError("ClickUp client not initialized")
workspaces = self.client.get_workspaces()
return f"Available workspaces:\n" + "\n".join([
f"- {ws.get('name', 'Unknown')} (ID: {ws.get('id', 'Unknown')})"
for ws in workspaces
])
elif path == "help":
return """ClickUp MCP Server Help
Available Tools:
- Task Management: create_task, get_task, update_task, delete_task, list_tasks, search_tasks
- Workspace Management: get_workspaces, get_workspace_members, get_spaces
- List Management: get_lists, create_list, update_list, delete_list
- Audit Logs: create_workspace_audit_log (Enterprise only)
Configuration:
Set CLICKUP_API_TOKEN environment variable with your ClickUp API token.
For detailed API documentation, visit: https://developer.clickup.com/
"""
else:
raise ValueError(f"Unknown resource path: {path}")
@self.server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
"""List available tools"""
if not self._tools_initialized():
await self._initialize_tools()
tools = []
tools.extend(self.task_tools.get_tools())
tools.extend(self.workspace_tools.get_tools())
tools.extend(self.list_tools.get_tools())
return tools
@self.server.call_tool()
async def handle_call_tool(
name: str, arguments: dict[str, Any] | None
) -> list[types.TextContent]:
"""Handle tool calls"""
if not self._tools_initialized():
await self._initialize_tools()
if arguments is None:
arguments = {}
# Route to appropriate tool handler
task_tool_names = {
"create_task", "get_task", "update_task", "delete_task",
"list_tasks", "search_tasks"
}
workspace_tool_names = {
"get_workspaces", "get_workspace_members", "get_spaces",
"create_workspace_audit_log"
}
list_tool_names = {
"get_lists", "create_list", "update_list", "delete_list"
}
try:
if name in task_tool_names:
return await self.task_tools.handle_tool_call(name, arguments)
elif name in workspace_tool_names:
return await self.workspace_tools.handle_tool_call(name, arguments)
elif name in list_tool_names:
return await self.list_tools.handle_tool_call(name, arguments)
else:
raise ValueError(f"Unknown tool: {name}")
except Exception as e:
return [types.TextContent(type="text", text=f"Error executing {name}: {str(e)}")]
def _tools_initialized(self) -> bool:
"""Check if tools are initialized"""
return (self.client is not None and
self.task_tools is not None and
self.workspace_tools is not None and
self.list_tools is not None)
async def _initialize_tools(self):
"""Initialize ClickUp client and tools"""
if not self.client:
self.client = ClickUpClient()
self.task_tools = TaskTools(self.client)
self.workspace_tools = WorkspaceTools(self.client)
self.list_tools = ListTools(self.client)
async def run(self):
"""Run the MCP server"""
# Initialize tools
await self._initialize_tools()
# Run server using stdio transport
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
await self.server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="clickup-mcp",
server_version="1.0.0",
capabilities=self.server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={},
),
),
)
async def main():
"""Main entry point"""
server = ClickUpMCPServer()
await server.run()