Implement: US: Authenticate and Connect to Trello Account #5

Merged
crisleo94 merged 2 commits from feature/trello-auth into main 2026-05-26 22:46:26 +00:00
3 changed files with 347 additions and 56 deletions
Showing only changes of commit 43acd5bfbf - Show all commits
+341 -46
View File
@@ -8,13 +8,47 @@ Provides the TrelloClient for interacting with the Trello REST API.
from __future__ import annotations
import os
from typing import Any
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.
@@ -23,6 +57,10 @@ class TrelloClient:
JSON serialization by Hermes tool handlers.
"""
api_key: str
token: str
_session: requests.Session
def __init__(
self,
api_key: str | None = None,
@@ -41,13 +79,42 @@ class TrelloClient:
"""Return query parameters common to every Trello API call."""
return {"key": self.api_key, "token": self.token}
def _get(self, path: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
"""Make an authenticated GET request and return the JSON response."""
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 {
"success": False,
"message": "TRELLO_API_KEY and TRELLO_TOKEN must both be set as environment variables.",
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()
@@ -55,46 +122,46 @@ class TrelloClient:
merged_params.update(params)
try:
resp = self._session.get(url, params=merged_params, timeout=15)
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 {"success": True, "data": data}
return SuccessResponse(success=True, data=data)
except requests.exceptions.HTTPError as exc:
status = exc.response.status_code if exc.response is not None else 0
if status == 401:
return {
"success": False,
"message": "Trello authentication failed (401). Check your TRELLO_API_KEY and TRELLO_TOKEN.",
}
if status == 403:
return {
"success": False,
"message": "Trello access denied (403). Your token may not have the required scopes.",
}
if status == 429:
return {
"success": False,
"message": "Trello rate limit exceeded. Try again later.",
}
return {
"success": False,
"message": f"Trello API error ({status}): {exc}",
}
return self._handle_http_error(exc)
except requests.exceptions.ConnectionError:
return {
"success": False,
"message": "Could not connect to Trello API. Check your network connection.",
}
return ErrorResponse(
success=False,
message="Could not connect to Trello API. Check your network connection.",
)
except requests.exceptions.Timeout:
return {
"success": False,
"message": "Trello API request timed out. Try again later.",
}
return ErrorResponse(
success=False,
message="Trello API request timed out. Try again later.",
)
except requests.exceptions.RequestException as exc:
return {
"success": False,
"message": f"Trello API request failed: {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
@@ -108,7 +175,7 @@ class TrelloClient:
``member`` key contains the Trello member id and username.
"""
result = self._get("/members/me")
if result["success"]:
if _is_success(result):
member = result["data"]
return {
"success": True,
@@ -134,8 +201,10 @@ class TrelloClient:
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 result["success"]:
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."),
@@ -156,6 +225,10 @@ class TrelloClient:
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.
"""
@@ -164,12 +237,234 @@ class TrelloClient:
return {
"success": True,
"message": (
"Trello credentials cleared. "
"Set TRELLO_API_KEY and TRELLO_TOKEN again to reconnect."
"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", "")}
# 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", [])
],
},
}
@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"))
return bool(os.environ.get("TRELLO_API_KEY")) and bool(
os.environ.get("TRELLO_TOKEN")
)
+1 -1
View File
@@ -73,7 +73,7 @@ def trello_disconnect() -> str:
"""
global _client # noqa: PLW0603
_client = None
return _respond({"success": True, "message": "Trello credentials cleared. Reconnect by calling trello_verify_credentials after setting TRELLO_API_KEY and TRELLO_TOKEN."})
return _respond({"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."})
# ---------------------------------------------------------------------------
+4 -8
View File
@@ -8,6 +8,7 @@ from collections.abc import Generator
from typing import Any
import pytest
from pytest import MonkeyPatch
import requests
# Tell pytest we plan to set env vars in tests
@@ -19,16 +20,11 @@ pytestmark = pytest.mark.usefixtures("clear_env")
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def clear_env() -> Generator[None, None, None]:
def clear_env(monkeypatch: MonkeyPatch) -> Generator[None, None, None]:
"""Remove Trello env vars before each test so state is predictable."""
# Use monkeypatch via request directly — fallback to os.environ
saved_key = os.environ.pop("TRELLO_API_KEY", None)
saved_token = os.environ.pop("TRELLO_TOKEN", None)
monkeypatch.delenv("TRELLO_API_KEY", raising=False)
monkeypatch.delenv("TRELLO_TOKEN", raising=False)
yield
if saved_key is not None:
os.environ["TRELLO_API_KEY"] = saved_key
if saved_token is not None:
os.environ["TRELLO_TOKEN"] = saved_token
def _make_session(mocker: Any) -> requests.Session: