"""
Trello MCP Server
An MCP server that provides tools for interacting with the Trello API using Nango authentication.
"""
import asyncio
import json
import logging
import sys
from typing import Any, Dict, List, Optional
import requests
from urllib.parse import urljoin
import os
from dotenv import load_dotenv
import mcp.types as types
from mcp.server import NotificationOptions, Server
from mcp.server.models import InitializationOptions
import mcp.server.stdio
# Load environment variables
load_dotenv(override=True)
# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("trello-mcp-server")
def get_connection_credentials(id: str, providerConfigKey: str) -> Dict[str, Any]:
"""Get credentials from Nango"""
base_url = os.environ.get("NANGO_BASE_URL")
secret_key = os.environ.get("NANGO_SECRET_KEY")
if not base_url or not secret_key:
raise ValueError("NANGO_BASE_URL and NANGO_SECRET_KEY environment variables are required")
url = f"{base_url}/connection/{id}"
params = {
"provider_config_key": providerConfigKey,
"refresh_token": "true",
}
headers = {"Authorization": f"Bearer {secret_key}"}
response = requests.get(url, headers=headers, params=params)
response.raise_for_status()
return response.json()
class TrelloAPIClient:
"""A Python client for the Trello REST API with Nango authentication."""
def __init__(self, connection_id: Optional[str] = None, integration_id: Optional[str] = None,
base_url: str = "https://api.trello.com/1/"):
self.connection_id = connection_id or os.environ.get("NANGO_CONNECTION_ID")
self.integration_id = integration_id or os.environ.get("NANGO_INTEGRATION_ID")
self.base_url = base_url
self.session = requests.Session()
if not self.connection_id:
raise ValueError("connection_id is required. Set NANGO_CONNECTION_ID environment variable.")
if not self.integration_id:
raise ValueError("integration_id is required. Set NANGO_INTEGRATION_ID environment variable.")
self._refresh_credentials()
self.session.headers.update({
'Accept': 'application/json',
'Content-Type': 'application/json'
})
def _refresh_credentials(self):
"""Refresh credentials from Nango and update session auth."""
try:
credentials = get_connection_credentials(self.connection_id, self.integration_id)
# Get API key from environment variable
self.api_key = os.environ.get('TRELLO_CLIENT_ID')
if not self.api_key:
raise ValueError("TRELLO_CLIENT_ID environment variable is required")
# Extract token from Nango response
if 'credentials' in credentials:
creds = credentials['credentials']
self.token = (
creds.get('oauth_token') or
creds.get('access_token') or
creds.get('token')
)
if not self.token:
raise ValueError(f"Missing required Trello token. Got credentials: {list(creds.keys())}")
else:
self.token = credentials.get('token') or credentials.get('access_token') or credentials.get('oauth_token')
if not self.token:
raise ValueError(f"Could not extract Trello token from Nango response: {list(credentials.keys())}")
# Update session with new credentials - KEY MUST COME FIRST!
session_params = {
'key': self.api_key,
'token': self.token
}
self.session.params.update(session_params)
logger.info(f"Successfully authenticated with Nango for connection: {self.connection_id}")
except Exception as e:
logger.error(f"Failed to get credentials from Nango: {e}")
raise
def _make_request(self, method: str, endpoint: str, params: Optional[Dict] = None,
data: Optional[Dict] = None, files: Optional[Dict] = None,
retry_on_auth_error: bool = True) -> Dict[Any, Any]:
"""Make a request to the Trello API with automatic credential refresh."""
url = urljoin(self.base_url, endpoint)
try:
if files:
headers = {'Accept': 'application/json'}
response = self.session.request(
method=method,
url=url,
params=params,
data=data,
files=files,
headers=headers
)
else:
response = self.session.request(
method=method,
url=url,
params=params,
json=data if data else None
)
response.raise_for_status()
return response.json() if response.content else {}
except requests.exceptions.HTTPError as e:
if e.response.status_code in [401, 403] and retry_on_auth_error:
logger.info("Authentication error detected, refreshing credentials...")
self._refresh_credentials()
return self._make_request(method, endpoint, params, data, files, retry_on_auth_error=False)
else:
logger.error(f"API Request failed: {e}")
if hasattr(e, 'response') and e.response is not None:
logger.error(f"Response: {e.response.text}")
raise
except requests.exceptions.RequestException as e:
logger.error(f"API Request failed: {e}")
raise
# API Methods
def get_board(self, board_id: str, **kwargs) -> Dict:
"""Get a board by ID."""
return self._make_request('GET', f'boards/{board_id}', params=kwargs)
def create_board(self, name: str, **kwargs) -> Dict:
"""Create a new board."""
params = {'name': name, **kwargs}
return self._make_request('POST', 'boards', params=params)
def update_board(self, board_id: str, **kwargs) -> Dict:
"""Update a board."""
return self._make_request('PUT', f'boards/{board_id}', params=kwargs)
def delete_board(self, board_id: str) -> Dict:
"""Delete a board."""
return self._make_request('DELETE', f'boards/{board_id}')
def get_board_lists(self, board_id: str, **kwargs) -> List[Dict]:
"""Get lists on a board."""
return self._make_request('GET', f'boards/{board_id}/lists', params=kwargs)
def get_board_cards(self, board_id: str, **kwargs) -> List[Dict]:
"""Get cards on a board."""
return self._make_request('GET', f'boards/{board_id}/cards', params=kwargs)
def get_board_members(self, board_id: str, **kwargs) -> List[Dict]:
"""Get members of a board."""
return self._make_request('GET', f'boards/{board_id}/members', params=kwargs)
def get_list(self, list_id: str, **kwargs) -> Dict:
"""Get a list by ID."""
return self._make_request('GET', f'lists/{list_id}', params=kwargs)
def create_list(self, name: str, board_id: str, **kwargs) -> Dict:
"""Create a new list on a board."""
params = {'name': name, 'idBoard': board_id, **kwargs}
return self._make_request('POST', 'lists', params=params)
def update_list(self, list_id: str, **kwargs) -> Dict:
"""Update a list."""
return self._make_request('PUT', f'lists/{list_id}', params=kwargs)
def archive_list(self, list_id: str) -> Dict:
"""Archive a list."""
return self._make_request('PUT', f'lists/{list_id}/closed', params={'value': 'true'})
def get_list_cards(self, list_id: str, **kwargs) -> List[Dict]:
"""Get cards in a list."""
return self._make_request('GET', f'lists/{list_id}/cards', params=kwargs)
def get_card(self, card_id: str, **kwargs) -> Dict:
"""Get a card by ID."""
return self._make_request('GET', f'cards/{card_id}', params=kwargs)
def create_card(self, name: str, list_id: str, **kwargs) -> Dict:
"""Create a new card."""
params = {'name': name, 'idList': list_id, **kwargs}
return self._make_request('POST', 'cards', params=params)
def update_card(self, card_id: str, **kwargs) -> Dict:
"""Update a card."""
return self._make_request('PUT', f'cards/{card_id}', params=kwargs)
def delete_card(self, card_id: str) -> Dict:
"""Delete a card."""
return self._make_request('DELETE', f'cards/{card_id}')
def add_comment_to_card(self, card_id: str, text: str) -> Dict:
"""Add a comment to a card."""
return self._make_request('POST', f'cards/{card_id}/actions/comments', params={'text': text})
def add_attachment_to_card(self, card_id: str, url: str = None, file_path: str = None, **kwargs) -> Dict:
"""Add an attachment to a card."""
if url:
params = {'url': url, **kwargs}
return self._make_request('POST', f'cards/{card_id}/attachments', params=params)
elif file_path:
with open(file_path, 'rb') as f:
files = {'file': f}
return self._make_request('POST', f'cards/{card_id}/attachments', files=files, data=kwargs)
else:
raise ValueError("Either url or file_path must be provided")
def get_member(self, member_id: str, **kwargs) -> Dict:
"""Get a member by ID or username."""
return self._make_request('GET', f'members/{member_id}', params=kwargs)
def get_my_boards(self, **kwargs) -> List[Dict]:
"""Get boards for the authenticated member."""
return self._make_request('GET', 'members/me/boards', params=kwargs)
def get_my_cards(self, **kwargs) -> List[Dict]:
"""Get cards for the authenticated member."""
return self._make_request('GET', 'members/me/cards', params=kwargs)
def get_organization(self, org_id: str, **kwargs) -> Dict:
"""Get an organization by ID."""
return self._make_request('GET', f'organizations/{org_id}', params=kwargs)
def create_organization(self, display_name: str, **kwargs) -> Dict:
"""Create a new organization."""
params = {'displayName': display_name, **kwargs}
return self._make_request('POST', 'organizations', params=params)
def search(self, query: str, **kwargs) -> Dict:
"""Search Trello."""
params = {'query': query, **kwargs}
return self._make_request('GET', 'search', params=params)
def search_members(self, query: str, **kwargs) -> List[Dict]:
"""Search for members."""
params = {'query': query, **kwargs}
return self._make_request('GET', 'search/members', params=params)
# Initialize the Trello client
trello_client = None
def get_trello_client() -> TrelloAPIClient:
"""Get or create the Trello client instance."""
global trello_client
if trello_client is None:
trello_client = TrelloAPIClient()
return trello_client
# Create the MCP server
server = Server("trello-server")
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
"""List available Trello tools."""
return [
types.Tool(
name="trello_get_my_boards",
description="Get all boards for the authenticated user",
inputSchema={
"type": "object",
"properties": {},
"additionalProperties": False,
},
),
types.Tool(
name="trello_get_board",
description="Get details of a specific board",
inputSchema={
"type": "object",
"properties": {
"board_id": {"type": "string", "description": "The ID of the board"},
},
"required": ["board_id"],
"additionalProperties": False,
},
),
types.Tool(
name="trello_create_board",
description="Create a new board",
inputSchema={
"type": "object",
"properties": {
"name": {"type": "string", "description": "The name of the board"},
"desc": {"type": "string", "description": "The description of the board"},
"defaultLists": {"type": "boolean", "description": "Whether to create default lists"},
},
"required": ["name"],
"additionalProperties": False,
},
),
types.Tool(
name="trello_update_board",
description="Update a board",
inputSchema={
"type": "object",
"properties": {
"board_id": {"type": "string", "description": "The ID of the board"},
"name": {"type": "string", "description": "The new name of the board"},
"desc": {"type": "string", "description": "The new description of the board"},
},
"required": ["board_id"],
"additionalProperties": False,
},
),
types.Tool(
name="trello_delete_board",
description="Delete a board",
inputSchema={
"type": "object",
"properties": {
"board_id": {"type": "string", "description": "The ID of the board"},
},
"required": ["board_id"],
"additionalProperties": False,
},
),
types.Tool(
name="trello_get_board_lists",
description="Get all lists on a board",
inputSchema={
"type": "object",
"properties": {
"board_id": {"type": "string", "description": "The ID of the board"},
},
"required": ["board_id"],
"additionalProperties": False,
},
),
types.Tool(
name="trello_get_board_cards",
description="Get all cards on a board",
inputSchema={
"type": "object",
"properties": {
"board_id": {"type": "string", "description": "The ID of the board"},
},
"required": ["board_id"],
"additionalProperties": False,
},
),
types.Tool(
name="trello_get_board_members",
description="Get all members of a board",
inputSchema={
"type": "object",
"properties": {
"board_id": {"type": "string", "description": "The ID of the board"},
},
"required": ["board_id"],
"additionalProperties": False,
},
),
types.Tool(
name="trello_create_list",
description="Create a new list on a board",
inputSchema={
"type": "object",
"properties": {
"name": {"type": "string", "description": "The name of the list"},
"board_id": {"type": "string", "description": "The ID of the board"},
"pos": {"type": "string", "description": "The position of the list (top, bottom, or number)"},
},
"required": ["name", "board_id"],
"additionalProperties": False,
},
),
types.Tool(
name="trello_update_list",
description="Update a list",
inputSchema={
"type": "object",
"properties": {
"list_id": {"type": "string", "description": "The ID of the list"},
"name": {"type": "string", "description": "The new name of the list"},
"pos": {"type": "string", "description": "The new position of the list"},
},
"required": ["list_id"],
"additionalProperties": False,
},
),
types.Tool(
name="trello_archive_list",
description="Archive a list",
inputSchema={
"type": "object",
"properties": {
"list_id": {"type": "string", "description": "The ID of the list"},
},
"required": ["list_id"],
"additionalProperties": False,
},
),
types.Tool(
name="trello_get_list_cards",
description="Get all cards in a list",
inputSchema={
"type": "object",
"properties": {
"list_id": {"type": "string", "description": "The ID of the list"},
},
"required": ["list_id"],
"additionalProperties": False,
},
),
types.Tool(
name="trello_create_card",
description="Create a new card",
inputSchema={
"type": "object",
"properties": {
"name": {"type": "string", "description": "The name of the card"},
"list_id": {"type": "string", "description": "The ID of the list"},
"desc": {"type": "string", "description": "The description of the card"},
"pos": {"type": "string", "description": "The position of the card (top, bottom, or number)"},
},
"required": ["name", "list_id"],
"additionalProperties": False,
},
),
types.Tool(
name="trello_get_card",
description="Get details of a specific card",
inputSchema={
"type": "object",
"properties": {
"card_id": {"type": "string", "description": "The ID of the card"},
},
"required": ["card_id"],
"additionalProperties": False,
},
),
types.Tool(
name="trello_update_card",
description="Update a card",
inputSchema={
"type": "object",
"properties": {
"card_id": {"type": "string", "description": "The ID of the card"},
"name": {"type": "string", "description": "The new name of the card"},
"desc": {"type": "string", "description": "The new description of the card"},
"idList": {"type": "string", "description": "Move card to different list"},
"pos": {"type": "string", "description": "The new position of the card"},
},
"required": ["card_id"],
"additionalProperties": False,
},
),
types.Tool(
name="trello_delete_card",
description="Delete a card",
inputSchema={
"type": "object",
"properties": {
"card_id": {"type": "string", "description": "The ID of the card"},
},
"required": ["card_id"],
"additionalProperties": False,
},
),
types.Tool(
name="trello_add_comment_to_card",
description="Add a comment to a card",
inputSchema={
"type": "object",
"properties": {
"card_id": {"type": "string", "description": "The ID of the card"},
"text": {"type": "string", "description": "The comment text"},
},
"required": ["card_id", "text"],
"additionalProperties": False,
},
),
types.Tool(
name="trello_add_attachment_to_card",
description="Add an attachment to a card",
inputSchema={
"type": "object",
"properties": {
"card_id": {"type": "string", "description": "The ID of the card"},
"url": {"type": "string", "description": "URL of the attachment"},
"name": {"type": "string", "description": "Name of the attachment"},
},
"required": ["card_id"],
"additionalProperties": False,
},
),
types.Tool(
name="trello_search",
description="Search Trello for boards, cards, etc.",
inputSchema={
"type": "object",
"properties": {
"query": {"type": "string", "description": "The search query"},
"modelTypes": {"type": "string", "description": "Comma-separated list of model types (boards,cards,members,organizations)"},
"boards_limit": {"type": "integer", "description": "Maximum number of boards to return"},
"cards_limit": {"type": "integer", "description": "Maximum number of cards to return"},
},
"required": ["query"],
"additionalProperties": False,
},
),
]
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict | None) -> list[types.TextContent]:
"""Handle tool calls."""
if arguments is None:
arguments = {}
try:
client = get_trello_client()
if name == "trello_get_my_boards":
result = client.get_my_boards()
elif name == "trello_get_board":
result = client.get_board(arguments["board_id"])
elif name == "trello_create_board":
result = client.create_board(**arguments)
elif name == "trello_update_board":
board_id = arguments.pop("board_id")
result = client.update_board(board_id, **arguments)
elif name == "trello_delete_board":
result = client.delete_board(arguments["board_id"])
elif name == "trello_get_board_lists":
result = client.get_board_lists(arguments["board_id"])
elif name == "trello_get_board_cards":
result = client.get_board_cards(arguments["board_id"])
elif name == "trello_get_board_members":
result = client.get_board_members(arguments["board_id"])
elif name == "trello_create_list":
name_arg = arguments.pop("name")
board_id = arguments.pop("board_id")
result = client.create_list(name_arg, board_id, **arguments)
elif name == "trello_update_list":
list_id = arguments.pop("list_id")
result = client.update_list(list_id, **arguments)
elif name == "trello_archive_list":
result = client.archive_list(arguments["list_id"])
elif name == "trello_get_list_cards":
result = client.get_list_cards(arguments["list_id"])
elif name == "trello_create_card":
name_arg = arguments.pop("name")
list_id = arguments.pop("list_id")
result = client.create_card(name_arg, list_id, **arguments)
elif name == "trello_get_card":
result = client.get_card(arguments["card_id"])
elif name == "trello_update_card":
card_id = arguments.pop("card_id")
result = client.update_card(card_id, **arguments)
elif name == "trello_delete_card":
result = client.delete_card(arguments["card_id"])
elif name == "trello_add_comment_to_card":
result = client.add_comment_to_card(arguments["card_id"], arguments["text"])
elif name == "trello_add_attachment_to_card":
card_id = arguments.pop("card_id")
result = client.add_attachment_to_card(card_id, **arguments)
elif name == "trello_search":
query = arguments.pop("query")
result = client.search(query, **arguments)
else:
raise ValueError(f"Unknown tool: {name}")
return [
types.TextContent(
type="text",
text=json.dumps(result, indent=2, default=str)
)
]
except Exception as e:
logger.error(f"Error in tool {name}: {e}")
return [
types.TextContent(
type="text",
text=f"Error: {str(e)}"
)
]
async def main():
"""Main function to run the MCP server."""
# Verify required environment variables
required_vars = ['NANGO_BASE_URL', 'NANGO_SECRET_KEY', 'NANGO_CONNECTION_ID', 'NANGO_INTEGRATION_ID', 'TRELLO_CLIENT_ID']
missing_vars = [var for var in required_vars if not os.getenv(var)]
if missing_vars:
logger.error(f"Missing required environment variables: {missing_vars}")
sys.exit(1)
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="trello-server",
server_version="0.1.0",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={},
),
),
)
def run():
asyncio.run(main())
