Taiga MCP Bridge

Instructions

Implementation Reference

• src/server.py:578-584 (handler)

  The handler function for the 'assign_task_to_user' MCP tool. It delegates the assignment by calling update_task with assigned_to=user_id.
  @mcp.tool("assign_task_to_user", description="Assigns a specific task to a specific user.") def assign_task_to_user(session_id: str, task_id: int, user_id: int) -> Dict[str, Any]: """Assigns a task to a user.""" logger.info( f"Executing assign_task_to_user: Task {task_id} -> User {user_id}, session {session_id[:8]}...") # Delegate to update_task return update_task(session_id, task_id, assigned_to=user_id)
• src/server.py:578-584 (registration)

  The @mcp.tool decorator registers the 'assign_task_to_user' tool with FastMCP.
  @mcp.tool("assign_task_to_user", description="Assigns a specific task to a specific user.") def assign_task_to_user(session_id: str, task_id: int, user_id: int) -> Dict[str, Any]: """Assigns a task to a user.""" logger.info( f"Executing assign_task_to_user: Task {task_id} -> User {user_id}, session {session_id[:8]}...") # Delegate to update_task return update_task(session_id, task_id, assigned_to=user_id)
• src/server.py:521-554 (helper)

  The update_task function performs the core logic of updating a task via the Taiga API (tasks.edit), including fetching version and handling assigned_to changes. It is called by assign_task_to_user.
  @mcp.tool("update_task", description="Updates details of an existing task.") def update_task(session_id: str, task_id: int, **kwargs) -> Dict[str, Any]: """Updates a task. Pass fields to update as keyword arguments (e.g., subject, description, status_id, assigned_to).""" logger.info( f"Executing update_task ID {task_id} for session {session_id[:8]} with data: {kwargs}") taiga_client_wrapper = _get_authenticated_client(session_id) # Use wrapper variable name try: # Use pytaigaclient edit pattern for partial updates if not kwargs: logger.info(f"No fields provided for update on task {task_id}") return taiga_client_wrapper.api.tasks.get(task_id) # Get current task data to retrieve version current_task = taiga_client_wrapper.api.tasks.get(task_id) version = current_task.get('version') if not version: raise ValueError(f"Could not determine version for task {task_id}") # Use edit method for partial updates with keyword arguments updated_task = taiga_client_wrapper.api.tasks.edit( task_id=task_id, version=version, **kwargs ) logger.info(f"Task {task_id} update request sent.") return updated_task except TaigaException as e: logger.error( f"Taiga API error updating task {task_id}: {e}", exc_info=False) raise e except Exception as e: logger.error( f"Unexpected error updating task {task_id}: {e}", exc_info=True) raise RuntimeError(f"Server error updating task: {e}")
• src/server.py:39-52 (helper)

  Helper function to retrieve and validate the authenticated Taiga client for a session, used by all tools including assign_task_to_user (indirectly).
  def _get_authenticated_client(session_id: str) -> TaigaClientWrapper: """ Retrieves the authenticated TaigaClientWrapper for a given session ID. Raises PermissionError if the session is invalid or not found. """ client = active_sessions.get(session_id) # Also check if the client object itself exists and is authenticated if not client or not client.is_authenticated: logger.warning(f"Invalid or expired session ID provided: {session_id}") # Raise PermissionError - FastMCP will map this to an appropriate error response raise PermissionError( f"Invalid or expired session ID: '{session_id}'. Please login again.") logger.debug(f"Retrieved valid client for session ID: {session_id}") return client