Taiga MCP Bridge

Instructions

Implementation Reference

• src/server.py:448-467 (handler)

  The handler function for the 'list_tasks' MCP tool. It retrieves tasks from a Taiga project using the authenticated client wrapper and supports optional filters.
  @mcp.tool("list_tasks", description="Lists tasks within a specific project, optionally filtered.") def list_tasks(session_id: str, project_id: int, **filters) -> List[Dict[str, Any]]: """Lists tasks for a project. Optional filters like 'milestone', 'status', 'user_story', 'assigned_to' can be passed as keyword arguments.""" logger.info( f"Executing list_tasks for project {project_id}, session {session_id[:8]}, filters: {filters}") taiga_client_wrapper = _get_authenticated_client(session_id) # Use wrapper variable name try: # Use pytaigaclient syntax: client.resource.list(project=..., **filters) tasks = taiga_client_wrapper.api.tasks.list(project=project_id, **filters) # return [t.to_dict() for t in tasks] # Remove .to_dict() return tasks # Return directly except TaigaException as e: logger.error( f"Taiga API error listing tasks for project {project_id}: {e}", exc_info=False) raise e except Exception as e: logger.error( f"Unexpected error listing tasks for project {project_id}: {e}", exc_info=True) raise RuntimeError(f"Server error listing tasks: {e}")
• src/server.py:39-52 (helper)

  Helper function used by list_tasks to get the authenticated Taiga client for the session.
  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
• src/taiga_client.py:15-80 (helper)

  The TaigaClientWrapper class provides the API client instance (api.tasks.list) used in the list_tasks handler.
  class TaigaClientWrapper: """ A wrapper around the pytaiga-client library to manage API instance and authentication state. """ def __init__(self, host: str): if not host: raise ValueError("Taiga host URL cannot be empty.") # Store host, but initialize client later during login/token auth self.host = host # Use the new client type self.api: Optional[TaigaClient] = None logger.info(f"TaigaClientWrapper initialized for host: {self.host}") def login(self, username: str, password: str) -> bool: """ Authenticates with the Taiga instance using username and password. Uses pytaigaclient. """ try: logger.info( f"Attempting login for user '{username}' on {self.host}") # Initialize the client here api_instance = TaigaClient(host=self.host) # Use the auth resource's login method api_instance.auth.login(username=username, password=password) self.api = api_instance logger.info( f"Login successful for user '{username}'. Auth token acquired.") return True except TaigaException as e: logger.error( f"Taiga login failed for user '{username}': {e}", exc_info=False) self.api = None raise e except Exception as e: logger.error( f"An unexpected error occurred during login for user '{username}': {e}", exc_info=True) self.api = None # Wrap unexpected errors in TaigaException if needed, or re-raise raise TaigaException(f"Unexpected login error: {e}") # Add method for token authentication if needed by pytaigaclient # def set_token(self, token: str, token_type: str = "Bearer"): # logger.info(f"Initializing TaigaClient with token on {self.host}") # self.api = TaigaClient(host=self.host, auth_token=token, token_type=token_type) # logger.info("TaigaClient initialized with token.") @property def is_authenticated(self) -> bool: """Checks if the client is currently authenticated (has an API instance with a token).""" # Check if api exists and has a token return self.api is not None and self.api.auth_token is not None def _ensure_authenticated(self): """Internal helper to check authentication before API calls.""" if not self.is_authenticated: logger.error( "Action required authentication, but client is not logged in.") # Use a standard exception type that FastMCP might handle better, # or a custom one if needed. PermissionError fits well. raise PermissionError( "Client not authenticated. Please login first.") # No changes needed to _ensure_authenticated or is_authenticated property logic,
• src/server.py:448-448 (registration)

  The @mcp.tool decorator registers the list_tasks function as an MCP tool.
  @mcp.tool("list_tasks", description="Lists tasks within a specific project, optionally filtered.")