Dooray MCP Server

Instructions

Implementation Reference

• src/dooray_mcp/server.py:150-228 (registration)

  Registers the dooray_search tool in the MCP list_tools handler, defining its name, description, and comprehensive input schema for different search types.

  types.Tool(
      name="dooray_search",
      description="Search Dooray content - tasks by various criteria including workflow, assignee, tags, status, date range with AND/OR logic",
      inputSchema={
          "type": "object",
          "properties": {
              "searchType": {
                  "type": "string",
                  "enum": ["tasks", "by_assignee", "by_status", "by_tag", "by_date_range", "by_workflow", "advanced"],
                  "description": "Type of search to perform"
              },
              "query": {
                  "type": "string",
                  "description": "Search query text (for tasks search)"
              },
              "assigneeId": {
                  "type": "string",
                  "description": "Assignee ID (for by_assignee search)"
              },
              "status": {
                  "type": "string",
                  "description": "Task status (for by_status search)"
              },
              "tagName": {
                  "type": "string",
                  "description": "Tag name (for by_tag search)"
              },
              "workflowId": {
                  "type": "string",
                  "description": "Workflow ID (for by_workflow search)"
              },
              "startDate": {
                  "type": "string",
                  "description": "Start date (for by_date_range search)"
              },
              "endDate": {
                  "type": "string",
                  "description": "End date (for by_date_range search)"
              },
              "conditions": {
                  "type": "array",
                  "items": {
                      "type": "object",
                      "properties": {
                          "type": {
                              "type": "string",
                              "enum": ["workflow", "assignee", "tag", "status", "query", "date_range"],
                              "description": "Type of search condition"
                          },
                          "value": {
                              "type": "string",
                              "description": "Value for the condition (workflow ID, assignee ID, tag name, status, or query text)"
                          },
                          "startDate": {
                              "type": "string",
                              "description": "Start date (for date_range type)"
                          },
                          "endDate": {
                              "type": "string",
                              "description": "End date (for date_range type)"
                          }
                      },
                      "required": ["type"]
                  },
                  "description": "Array of search conditions (for advanced search)"
              },
              "logicOperator": {
                  "type": "string",
                  "enum": ["AND", "OR"],
                  "description": "Logic operator for combining conditions in advanced search (default: AND)"
              },
              "limit": {
                  "type": "integer",
                  "description": "Maximum results to return (optional)"
              }
          },
          "required": ["searchType"]
      }
  ),

• src/dooray_mcp/tools/search.py:9-163 (handler)

  The SearchTool class is the primary handler for dooray_search tool calls. Its handle method routes based on searchType to specialized methods that validate args, build params, call DoorayClient search APIs, and return JSON results.

  class SearchTool:
      """Tool for searching Dooray content."""
      
      def __init__(self, dooray_client):
          """Initialize with Dooray client."""
          self.client = dooray_client
      
      async def handle(self, arguments: Dict[str, Any]) -> str:
          """Handle search tool requests.
          
          Args:
              arguments: Tool arguments containing searchType and parameters
              
          Returns:
              JSON string with results
          """
          search_type = arguments.get("searchType")
          if not search_type:
              return json.dumps({"error": "searchType parameter is required"})
          
          try:
              if search_type == "tasks":
                  return await self._search_tasks(arguments)
              elif search_type == "by_assignee":
                  return await self._search_by_assignee(arguments)
              elif search_type == "by_status":
                  return await self._search_by_status(arguments)
              elif search_type == "by_tag":
                  return await self._search_by_tag(arguments)
              elif search_type == "by_date_range":
                  return await self._search_by_date_range(arguments)
              elif search_type == "by_workflow":
                  return await self._search_by_workflow(arguments)
              elif search_type == "advanced":
                  return await self._advanced_search(arguments)
              else:
                  return json.dumps({"error": f"Unknown search type: {search_type}"})
                  
          except Exception as e:
              logger.error(f"Error in search tool: {e}")
              return json.dumps({"error": str(e)})
      
      async def _search_tasks(self, arguments: Dict[str, Any]) -> str:
          """Search tasks by query text."""
          project_id = arguments.get("projectId")
          query = arguments.get("query")
          
          if not query:
              return json.dumps({"error": "query is required for tasks search"})
          
          # Build search parameters
          params = {}
          if arguments.get("limit"):
              params["size"] = arguments["limit"]
          
          result = await self.client.search_tasks(project_id, query, **params)
          return json.dumps(result, ensure_ascii=False)
      
      async def _search_by_assignee(self, arguments: Dict[str, Any]) -> str:
          """Search tasks by assignee."""
          project_id = arguments.get("projectId")
          assignee_id = arguments.get("assigneeId")
          
          if not assignee_id:
              return json.dumps({"error": "assigneeId is required for by_assignee search"})
          
          # Build search parameters
          params = {}
          if arguments.get("limit"):
              params["size"] = arguments["limit"]
          
          result = await self.client.search_tasks_by_assignee(project_id, assignee_id, **params)
          return json.dumps(result, ensure_ascii=False)
      
      async def _search_by_status(self, arguments: Dict[str, Any]) -> str:
          """Search tasks by status."""
          project_id = arguments.get("projectId")
          status = arguments.get("status")
          
          if not status:
              return json.dumps({"error": "status is required for by_status search"})
          
          # Build search parameters
          params = {}
          if arguments.get("limit"):
              params["size"] = arguments["limit"]
          
          result = await self.client.search_tasks_by_status(project_id, status, **params)
          return json.dumps(result, ensure_ascii=False)
      
      async def _search_by_tag(self, arguments: Dict[str, Any]) -> str:
          """Search tasks by tag."""
          project_id = arguments.get("projectId")
          tag_name = arguments.get("tagName")
          
          if not tag_name:
              return json.dumps({"error": "tagName is required for by_tag search"})
          
          # Build search parameters
          params = {}
          if arguments.get("limit"):
              params["size"] = arguments["limit"]
          
          result = await self.client.search_tasks_by_tag(project_id, tag_name, **params)
          return json.dumps(result, ensure_ascii=False)
      
      async def _search_by_date_range(self, arguments: Dict[str, Any]) -> str:
          """Search tasks by date range."""
          project_id = arguments.get("projectId")
          start_date = arguments.get("startDate")
          end_date = arguments.get("endDate")
          
          if not start_date or not end_date:
              return json.dumps({"error": "startDate and endDate are required for by_date_range search"})
          
          # Build search parameters
          params = {}
          if arguments.get("limit"):
              params["size"] = arguments["limit"]
          
          result = await self.client.search_tasks_by_date_range(project_id, start_date, end_date, **params)
          return json.dumps(result, ensure_ascii=False)
      
      async def _search_by_workflow(self, arguments: Dict[str, Any]) -> str:
          """Search tasks by workflow ID."""
          project_id = arguments.get("projectId")
          workflow_id = arguments.get("workflowId")
          
          if not workflow_id:
              return json.dumps({"error": "workflowId is required for by_workflow search"})
          
          # Build search parameters
          params = {}
          if arguments.get("limit"):
              params["size"] = arguments["limit"]
          
          result = await self.client.search_tasks_by_workflow(project_id, workflow_id, **params)
          return json.dumps(result, ensure_ascii=False)
      
      async def _advanced_search(self, arguments: Dict[str, Any]) -> str:
          """Advanced search with multiple conditions and AND/OR logic."""
          project_id = arguments.get("projectId")
          conditions = arguments.get("conditions", [])
          logic_operator = arguments.get("logicOperator", "AND")  # AND or OR
          
          if not conditions:
              return json.dumps({"error": "conditions array is required for advanced search"})
          
          # Build search parameters
          params = {}
          if arguments.get("limit"):
              params["size"] = arguments["limit"]
          
          result = await self.client.advanced_search_tasks(project_id, conditions, logic_operator, **params)
          return json.dumps(result, ensure_ascii=False)

• src/dooray_mcp/server.py:364-366 (registration)

  In the MCP server's call_tool handler, dooray_search requests are routed by instantiating SearchTool and invoking its handle method.

  elif name == "dooray_search":
      tool = SearchTool(dooray_client)
      result = await tool.handle(args)