""" Trello Plugin — Hermes Agent plugin for Trello board integration. Authentication & Connection module. Provides the TrelloClient for interacting with the Trello REST API. """ from __future__ import annotations import os from typing import Any, TypeGuard, TypedDict import requests # Base URL for the Trello REST API v1. # See https://developer.atlassian.com/cloud/trello/rest TRELLO_API_BASE = "https://api.trello.com/1" # --------------------------------------------------------------------------- # Typed response contracts # --------------------------------------------------------------------------- class SuccessResponse(TypedDict): """A successful API response carrying data.""" success: bool data: dict[str, Any] class ErrorResponse(TypedDict): """An unsuccessful API response with a human-readable message.""" success: bool message: str ApiResponse = SuccessResponse | ErrorResponse def _is_success(resp: ApiResponse) -> TypeGuard[SuccessResponse]: """Narrow an ApiResponse union to its success branch.""" return bool(resp.get("success", False)) # noqa: FBT003 # --------------------------------------------------------------------------- # Client # --------------------------------------------------------------------------- class TrelloClient: """Client for the Trello REST API. Reads credentials from environment variables TRELLO_API_KEY and TRELLO_TOKEN unless explicitly passed. All public methods return plain dicts suitable for JSON serialization by Hermes tool handlers. """ api_key: str token: str _session: requests.Session def __init__( self, api_key: str | None = None, token: str | None = None, session: requests.Session | None = None, ) -> None: self.api_key = api_key or os.environ.get("TRELLO_API_KEY", "") self.token = token or os.environ.get("TRELLO_TOKEN", "") self._session = session or requests.Session() # ------------------------------------------------------------------ # Internal helpers # ------------------------------------------------------------------ def _auth_params(self) -> dict[str, str]: """Return query parameters common to every Trello API call.""" return {"key": self.api_key, "token": self.token} def _check_credentials(self) -> ErrorResponse | None: """Return an ErrorResponse if credentials are missing, else None.""" if not self.api_key or not self.token: return ErrorResponse( success=False, message="TRELLO_API_KEY and TRELLO_TOKEN must both be set as environment variables.", ) return None def _handle_http_error(self, exc: requests.exceptions.HTTPError) -> ErrorResponse: """Map known HTTP status codes to user-facing error messages.""" status = exc.response.status_code if exc.response is not None else 0 messages: dict[int, str] = { 401: "Trello authentication failed (401). Check your TRELLO_API_KEY and TRELLO_TOKEN.", 403: "Trello access denied (403). Your token may not have the required scopes.", 429: "Trello rate limit exceeded. Try again later.", } message = messages.get( status, f"Trello API error ({status}): {exc}" ) return ErrorResponse(success=False, message=message) def _request( self, method: str, path: str, params: dict[str, Any] | None = None, json_body: dict[str, Any] | None = None, ) -> ApiResponse: """Make an authenticated HTTP request and return the JSON response. Handles credential checks, HTTP errors, network errors, and timeouts. """ cred_error = self._check_credentials() if cred_error is not None: return cred_error url = f"{TRELLO_API_BASE}{path}" merged_params = self._auth_params() if params: merged_params.update(params) try: resp = self._session.request( method, url, params=merged_params, json=json_body, timeout=15 ) resp.raise_for_status() data: dict[str, Any] = resp.json() return SuccessResponse(success=True, data=data) except requests.exceptions.HTTPError as exc: return self._handle_http_error(exc) except requests.exceptions.ConnectionError: return ErrorResponse( success=False, message="Could not connect to Trello API. Check your network connection.", ) except requests.exceptions.Timeout: return ErrorResponse( success=False, message="Trello API request timed out. Try again later.", ) except requests.exceptions.RequestException as exc: return ErrorResponse( success=False, message=f"Trello API request failed: {exc}" ) def _get( self, path: str, params: dict[str, Any] | None = None ) -> ApiResponse: """Make an authenticated GET request.""" return self._request("GET", path, params=params) def _post( self, path: str, json_body: dict[str, Any] | None = None ) -> ApiResponse: """Make an authenticated POST request.""" return self._request("POST", path, json_body=json_body) def _put( self, path: str, json_body: dict[str, Any] | None = None ) -> ApiResponse: """Make an authenticated PUT request.""" return self._request("PUT", path, json_body=json_body) # ------------------------------------------------------------------ # Public API — Auth & Connection # ------------------------------------------------------------------ def verify_credentials(self) -> dict[str, Any]: """Verify that the stored credentials are valid by fetching member info. Returns: dict with ``success`` bool and ``message`` string. On success the ``member`` key contains the Trello member id and username. """ result = self._get("/members/me") if _is_success(result): member = result["data"] return { "success": True, "message": ( f"✓ Trello credentials verified successfully " f"(authenticated as {member.get('username', 'unknown')})" ), "member": { "id": member.get("id"), "username": member.get("username"), "full_name": member.get("fullName"), }, } return { "success": False, "message": result.get("message", "Trello credential verification failed."), } def list_boards(self) -> dict[str, Any]: """Fetch all boards accessible to the authenticated user. Returns: dict with ``success`` bool and a ``boards`` list (each containing ``id``, ``name``, ``url``, ``closed``, ``starred``). """ result = self._get( "/members/me/boards", params={"fields": "id,name,url,closed,starred"} ) if not _is_success(result): return { "success": False, "message": result.get("message", "Failed to list Trello boards."), } boards = [ { "id": b.get("id"), "name": b.get("name"), "url": b.get("url"), "closed": b.get("closed", False), "starred": b.get("starred", False), } for b in result["data"] ] return {"success": True, "boards": boards} def disconnect(self) -> dict[str, Any]: """Clear credentials from the in-memory client. Since credentials are stored in environment variables, the next tool call will automatically re-read them and reconnect. To fully disconnect, also unset TRELLO_API_KEY and TRELLO_TOKEN from the Hermes profile config. Note: this does NOT revoke the Trello token. The user must invalidate it via Trello's account settings if they want full revocation. """ self.api_key = "" self.token = "" return { "success": True, "message": ( "Trello credentials cleared from memory. " "Next tool call will re-read TRELLO_API_KEY and TRELLO_TOKEN " "from environment and reconnect automatically." ), } # ------------------------------------------------------------------ # Public API — Board Management # ------------------------------------------------------------------ def _resolve_board_id(self, board_id: str) -> dict[str, Any]: """Resolve a board name or ID to a Trello board ID. Returns board info dict with at least an ``id`` key on success, or an error dict with ``success: False``. """ # Try as ID first (Trello IDs are 8-24 char hex strings) result = self._get(f"/boards/{board_id}", params={"fields": "id,name,url"}) if _is_success(result): board = result["data"] return {"id": board["id"], "name": board.get("name", ""), "url": board.get("url", "")} # If the error is about missing credentials, propagate that directly error_msg = result.get("message", "") if "TRELLO_API_KEY" in error_msg: return {"success": False, "message": error_msg} # Try resolving by name boards_result = self.list_boards() if not _is_success(boards_result): # type: ignore[arg-type] return { "success": False, "message": ( f"Board '{board_id}' not found. " "Use trello_list_boards to see available boards." ), } matches = [b for b in boards_result["boards"] if b["name"] == board_id] # type: ignore[typeddict-item] if len(matches) == 1: return dict(matches[0]) # already has id, name, url if len(matches) > 1: return { "success": False, "message": ( f"Multiple boards named '{board_id}' found. " "Use the board ID instead of the name." ), } return { "success": False, "message": ( f"Board '{board_id}' not found. " "Use trello_list_boards to see available boards." ), } def create_board( self, name: str, default_lists: bool = True ) -> dict[str, Any]: """Create a new Trello board. Args: name: The name for the new board. default_lists: Whether to create the default lists (default: True). Returns: dict with board details on success, or an error dict. """ body: dict[str, Any] = { "name": name, "defaultLists": default_lists, } result = self._post("/boards", json_body=body) if _is_success(result): board = result["data"] return { "success": True, "board": { "id": board.get("id"), "name": board.get("name"), "url": board.get("url"), }, } return { "success": False, "message": result.get("message", "Failed to create board."), } def rename_board(self, board_id: str, name: str) -> dict[str, Any]: """Rename an existing Trello board. Args: board_id: Board ID or name. name: The new name for the board. Returns: dict with board details on success, or an error dict. """ resolved = self._resolve_board_id(board_id) if "success" in resolved and resolved["success"] is False: return resolved # type: ignore[typeddict-item] trello_id = resolved["id"] result = self._put( f"/boards/{trello_id}", json_body={"name": name} ) if _is_success(result): board = result["data"] return { "success": True, "board": { "id": board.get("id"), "name": board.get("name"), "url": board.get("url"), }, } return { "success": False, "message": result.get("message", "Failed to rename board."), } def archive_board(self, board_id: str) -> dict[str, Any]: """Close/archive a Trello board. Args: board_id: Board ID or name. Returns: dict with success message or error. """ resolved = self._resolve_board_id(board_id) if "success" in resolved and resolved["success"] is False: return resolved # type: ignore[typeddict-item] trello_id = resolved["id"] board_name = resolved.get("name", trello_id) result = self._put( f"/boards/{trello_id}", json_body={"closed": True} ) if _is_success(result): return {"success": True, "message": f"Board '{board_name}' archived."} return { "success": False, "message": result.get("message", "Failed to archive board."), } def open_board(self, board_id: str) -> dict[str, Any]: """Re-open a closed/archived Trello board. Args: board_id: Board ID or name. Returns: dict with success message or error. """ resolved = self._resolve_board_id(board_id) if "success" in resolved and resolved["success"] is False: return resolved # type: ignore[typeddict-item] trello_id = resolved["id"] board_name = resolved.get("name", trello_id) result = self._put( f"/boards/{trello_id}", json_body={"closed": False} ) if _is_success(result): return {"success": True, "message": f"Board '{board_name}' opened."} return { "success": False, "message": result.get("message", "Failed to open board."), } def board_details(self, board_id: str) -> dict[str, Any]: """Fetch detailed information about a Trello board. Returns the board's lists, members, and a direct URL. Args: board_id: Board ID or name. Returns: dict with board details including lists and members on success. """ resolved = self._resolve_board_id(board_id) if "success" in resolved and resolved["success"] is False: return resolved # type: ignore[typeddict-item] trello_id = resolved["id"] result = self._get( f"/boards/{trello_id}", params={ "fields": "id,name,desc,url,closed,starred", "lists": "all", "members": "all", "members_fields": "id,username,fullName", }, ) if not _is_success(result): return { "success": False, "message": result.get("message", "Failed to fetch board details."), } board = result["data"] return { "success": True, "board": { "id": board.get("id"), "name": board.get("name"), "url": board.get("url"), "desc": board.get("desc", ""), "closed": board.get("closed", False), "starred": board.get("starred", False), "lists": [ {"id": lst.get("id"), "name": lst.get("name")} for lst in board.get("lists", []) ], "members": [ { "id": m.get("id"), "username": m.get("username"), "full_name": m.get("fullName"), } for m in board.get("members", []) ], }, } # ------------------------------------------------------------------ # Public API — List Management # ------------------------------------------------------------------ def create_list(self, name: str, board_id: str, pos: str = "bottom") -> dict[str, Any]: """Create a new list on a Trello board. Args: name: The name for the new list. board_id: Board ID or name. pos: Position — ``"top"``, ``"bottom"``, or a number. Returns: dict with list details on success, or an error dict. """ resolved = self._resolve_board_id(board_id) if "success" in resolved and resolved["success"] is False: return resolved # type: ignore[typeddict-item] body: dict[str, Any] = { "name": name, "idBoard": resolved["id"], "pos": pos, } result = self._post("/lists", json_body=body) if _is_success(result): lst = result["data"] return { "success": True, "list": { "id": lst.get("id"), "name": lst.get("name"), "id_board": lst.get("idBoard"), }, } return { "success": False, "message": result.get("message", "Failed to create list."), } def rename_list(self, list_id: str, name: str) -> dict[str, Any]: """Rename an existing list. Args: list_id: Trello list ID. name: The new name. Returns: dict with list details on success, or an error dict. """ result = self._put(f"/lists/{list_id}", json_body={"name": name}) if _is_success(result): lst = result["data"] return { "success": True, "list": { "id": lst.get("id"), "name": lst.get("name"), }, } return { "success": False, "message": result.get("message", "Failed to rename list."), } def archive_list(self, list_id: str) -> dict[str, Any]: """Archive a list. Args: list_id: Trello list ID. Returns: dict with success message or error. """ result = self._put(f"/lists/{list_id}", json_body={"closed": True}) if _is_success(result): lst = result["data"] list_name = lst.get("name", list_id) return {"success": True, "message": f"List '{list_name}' archived."} return { "success": False, "message": result.get("message", "Failed to archive list."), } def move_list(self, list_id: str, pos: str = "bottom") -> dict[str, Any]: """Move a list to a different position on the board. Args: list_id: Trello list ID. pos: Position — ``"top"``, ``"bottom"``, or a number. Returns: dict with success message or error. """ result = self._put(f"/lists/{list_id}", json_body={"pos": pos}) if _is_success(result): lst = result["data"] list_name = lst.get("name", list_id) return {"success": True, "message": f"List '{list_name}' moved."} return { "success": False, "message": result.get("message", "Failed to move list."), } @staticmethod def check_requirements() -> bool: """Check if the required environment variables are set.""" return bool(os.environ.get("TRELLO_API_KEY")) and bool( os.environ.get("TRELLO_TOKEN") )