🚀 MAJOR: Migrate to official FastMCP framework
- Create new basecamp_fastmcp.py using official Anthropic MCP SDK - Migrate 19 essential tools following official MCP best practices - Maintain identical tool names and API compatibility with original server - Add FastMCP dependencies (mcp[cli]>=1.2.0, httpx, anyio) - Use async/await with anyio.to_thread for sync→async bridge - Follow official logging practices (stderr + file, not stdout) - Full compatibility with Cursor and Claude Desktop Core tools migrated: ✅ Projects: get_projects, get_project, get_todolists, get_todos ✅ Search: search_basecamp, global_search ✅ Communication: get_comments, get_campfire_lines ✅ Card Tables: get_card_tables, get_card_table, get_columns, get_cards ✅ Card Management: create_card, get_card, update_card, move_card, complete_card ✅ Column Management: get_column, create_column Next: Update config generation for dual-run support
This commit is contained in:
George Antonopoulos
703
basecamp_fastmcp.py
Normal file
703
basecamp_fastmcp.py
Normal file
@@ -0,0 +1,703 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
FastMCP server for Basecamp integration.
|
||||
|
||||
This server implements the MCP (Model Context Protocol) using the official
|
||||
Anthropic FastMCP framework, replacing the custom JSON-RPC implementation.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
from typing import Any, Dict, List, Optional
|
||||
import anyio
|
||||
import httpx
|
||||
from mcp.server.fastmcp import FastMCP
|
||||
|
||||
# Import existing business logic
|
||||
from basecamp_client import BasecampClient
|
||||
from search_utils import BasecampSearch
|
||||
import token_storage
|
||||
from dotenv import load_dotenv
|
||||
|
||||
# Determine project root (directory containing this script)
|
||||
PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))
|
||||
DOTENV_PATH = os.path.join(PROJECT_ROOT, '.env')
|
||||
load_dotenv(DOTENV_PATH)
|
||||
|
||||
# Set up logging to file AND stderr (following MCP best practices)
|
||||
LOG_FILE_PATH = os.path.join(PROJECT_ROOT, 'basecamp_fastmcp.log')
|
||||
logging.basicConfig(
|
||||
level=logging.DEBUG,
|
||||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
|
||||
handlers=[
|
||||
logging.FileHandler(LOG_FILE_PATH),
|
||||
logging.StreamHandler(sys.stderr) # Critical: log to stderr, not stdout
|
||||
]
|
||||
)
|
||||
logger = logging.getLogger('basecamp_fastmcp')
|
||||
|
||||
# Initialize FastMCP server
|
||||
mcp = FastMCP("basecamp")
|
||||
|
||||
# Auth helper functions (reused from original server)
|
||||
def _get_basecamp_client() -> Optional[BasecampClient]:
|
||||
"""Get authenticated Basecamp client (sync version from original server)."""
|
||||
try:
|
||||
token_data = token_storage.get_token()
|
||||
logger.debug(f"Token data retrieved: {token_data}")
|
||||
|
||||
if not token_data or not token_data.get('access_token'):
|
||||
logger.error("No OAuth token available")
|
||||
return None
|
||||
|
||||
# Check if token is expired
|
||||
if token_storage.is_token_expired():
|
||||
logger.error("OAuth token has expired")
|
||||
return None
|
||||
|
||||
# Get account_id from token data first, then fall back to env var
|
||||
account_id = token_data.get('account_id') or os.getenv('BASECAMP_ACCOUNT_ID')
|
||||
user_agent = os.getenv('USER_AGENT') or "Basecamp MCP Server (cursor@example.com)"
|
||||
|
||||
if not account_id:
|
||||
logger.error(f"Missing account_id. Token data: {token_data}, Env BASECAMP_ACCOUNT_ID: {os.getenv('BASECAMP_ACCOUNT_ID')}")
|
||||
return None
|
||||
|
||||
logger.debug(f"Creating Basecamp client with account_id: {account_id}, user_agent: {user_agent}")
|
||||
|
||||
return BasecampClient(
|
||||
access_token=token_data['access_token'],
|
||||
account_id=account_id,
|
||||
user_agent=user_agent,
|
||||
auth_mode='oauth'
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Error creating Basecamp client: {e}")
|
||||
return None
|
||||
|
||||
def _get_auth_error_response() -> Dict[str, Any]:
|
||||
"""Return consistent auth error response."""
|
||||
if token_storage.is_token_expired():
|
||||
return {
|
||||
"error": "OAuth token expired",
|
||||
"message": "Your Basecamp OAuth token has expired. Please re-authenticate by visiting http://localhost:8000 and completing the OAuth flow again."
|
||||
}
|
||||
else:
|
||||
return {
|
||||
"error": "Authentication required",
|
||||
"message": "Please authenticate with Basecamp first. Visit http://localhost:8000 to log in."
|
||||
}
|
||||
|
||||
async def _run_sync(func, *args, **kwargs):
|
||||
"""Wrapper to run synchronous functions in thread pool."""
|
||||
return await anyio.to_thread.run_sync(func, *args, **kwargs)
|
||||
|
||||
# Core MCP Tools - Starting with essential ones from original server
|
||||
|
||||
@mcp.tool()
|
||||
async def get_projects() -> Dict[str, Any]:
|
||||
"""Get all Basecamp projects."""
|
||||
client = _get_basecamp_client()
|
||||
if not client:
|
||||
return _get_auth_error_response()
|
||||
|
||||
try:
|
||||
projects = await _run_sync(client.get_projects)
|
||||
return {
|
||||
"status": "success",
|
||||
"projects": projects,
|
||||
"count": len(projects)
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting projects: {e}")
|
||||
if "401" in str(e) and "expired" in str(e).lower():
|
||||
return {
|
||||
"error": "OAuth token expired",
|
||||
"message": "Your Basecamp OAuth token expired during the API call. Please re-authenticate by visiting http://localhost:8000 and completing the OAuth flow again."
|
||||
}
|
||||
return {
|
||||
"error": "Execution error",
|
||||
"message": str(e)
|
||||
}
|
||||
|
||||
@mcp.tool()
|
||||
async def get_project(project_id: str) -> Dict[str, Any]:
|
||||
"""Get details for a specific project.
|
||||
|
||||
Args:
|
||||
project_id: The project ID
|
||||
"""
|
||||
client = _get_basecamp_client()
|
||||
if not client:
|
||||
return _get_auth_error_response()
|
||||
|
||||
try:
|
||||
project = await _run_sync(client.get_project, project_id)
|
||||
return {
|
||||
"status": "success",
|
||||
"project": project
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting project {project_id}: {e}")
|
||||
if "401" in str(e) and "expired" in str(e).lower():
|
||||
return {
|
||||
"error": "OAuth token expired",
|
||||
"message": "Your Basecamp OAuth token expired during the API call. Please re-authenticate by visiting http://localhost:8000 and completing the OAuth flow again."
|
||||
}
|
||||
return {
|
||||
"error": "Execution error",
|
||||
"message": str(e)
|
||||
}
|
||||
|
||||
@mcp.tool()
|
||||
async def search_basecamp(query: str, project_id: Optional[str] = None) -> Dict[str, Any]:
|
||||
"""Search across Basecamp projects, todos, and messages.
|
||||
|
||||
Args:
|
||||
query: Search query
|
||||
project_id: Optional project ID to limit search scope
|
||||
"""
|
||||
client = _get_basecamp_client()
|
||||
if not client:
|
||||
return _get_auth_error_response()
|
||||
|
||||
try:
|
||||
search = BasecampSearch(client=client)
|
||||
results = {}
|
||||
|
||||
if project_id:
|
||||
# Search within specific project
|
||||
results["todolists"] = await _run_sync(search.search_todolists, query, project_id)
|
||||
results["todos"] = await _run_sync(search.search_todos, query, project_id)
|
||||
else:
|
||||
# Search across all projects
|
||||
results["projects"] = await _run_sync(search.search_projects, query)
|
||||
results["todos"] = await _run_sync(search.search_todos, query)
|
||||
results["messages"] = await _run_sync(search.search_messages, query)
|
||||
|
||||
return {
|
||||
"status": "success",
|
||||
"query": query,
|
||||
"results": results
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"Error searching Basecamp: {e}")
|
||||
if "401" in str(e) and "expired" in str(e).lower():
|
||||
return {
|
||||
"error": "OAuth token expired",
|
||||
"message": "Your Basecamp OAuth token expired during the API call. Please re-authenticate by visiting http://localhost:8000 and completing the OAuth flow again."
|
||||
}
|
||||
return {
|
||||
"error": "Execution error",
|
||||
"message": str(e)
|
||||
}
|
||||
|
||||
@mcp.tool()
|
||||
async def get_todolists(project_id: str) -> Dict[str, Any]:
|
||||
"""Get todo lists for a project.
|
||||
|
||||
Args:
|
||||
project_id: The project ID
|
||||
"""
|
||||
client = _get_basecamp_client()
|
||||
if not client:
|
||||
return _get_auth_error_response()
|
||||
|
||||
try:
|
||||
todolists = await _run_sync(client.get_todolists, project_id)
|
||||
return {
|
||||
"status": "success",
|
||||
"todolists": todolists,
|
||||
"count": len(todolists)
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting todolists: {e}")
|
||||
if "401" in str(e) and "expired" in str(e).lower():
|
||||
return {
|
||||
"error": "OAuth token expired",
|
||||
"message": "Your Basecamp OAuth token expired during the API call. Please re-authenticate by visiting http://localhost:8000 and completing the OAuth flow again."
|
||||
}
|
||||
return {
|
||||
"error": "Execution error",
|
||||
"message": str(e)
|
||||
}
|
||||
|
||||
@mcp.tool()
|
||||
async def get_todos(project_id: str, todolist_id: str) -> Dict[str, Any]:
|
||||
"""Get todos from a todo list.
|
||||
|
||||
Args:
|
||||
project_id: Project ID
|
||||
todolist_id: The todo list ID
|
||||
"""
|
||||
client = _get_basecamp_client()
|
||||
if not client:
|
||||
return _get_auth_error_response()
|
||||
|
||||
try:
|
||||
todos = await _run_sync(client.get_todos, project_id, todolist_id)
|
||||
return {
|
||||
"status": "success",
|
||||
"todos": todos,
|
||||
"count": len(todos)
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting todos: {e}")
|
||||
if "401" in str(e) and "expired" in str(e).lower():
|
||||
return {
|
||||
"error": "OAuth token expired",
|
||||
"message": "Your Basecamp OAuth token expired during the API call. Please re-authenticate by visiting http://localhost:8000 and completing the OAuth flow again."
|
||||
}
|
||||
return {
|
||||
"error": "Execution error",
|
||||
"message": str(e)
|
||||
}
|
||||
|
||||
@mcp.tool()
|
||||
async def global_search(query: str) -> Dict[str, Any]:
|
||||
"""Search projects, todos and campfire messages across all projects.
|
||||
|
||||
Args:
|
||||
query: Search query
|
||||
"""
|
||||
client = _get_basecamp_client()
|
||||
if not client:
|
||||
return _get_auth_error_response()
|
||||
|
||||
try:
|
||||
search = BasecampSearch(client=client)
|
||||
results = await _run_sync(search.global_search, query)
|
||||
return {
|
||||
"status": "success",
|
||||
"query": query,
|
||||
"results": results
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"Error in global search: {e}")
|
||||
if "401" in str(e) and "expired" in str(e).lower():
|
||||
return {
|
||||
"error": "OAuth token expired",
|
||||
"message": "Your Basecamp OAuth token expired during the API call. Please re-authenticate by visiting http://localhost:8000 and completing the OAuth flow again."
|
||||
}
|
||||
return {
|
||||
"error": "Execution error",
|
||||
"message": str(e)
|
||||
}
|
||||
|
||||
@mcp.tool()
|
||||
async def get_comments(recording_id: str, project_id: str) -> Dict[str, Any]:
|
||||
"""Get comments for a Basecamp item.
|
||||
|
||||
Args:
|
||||
recording_id: The item ID
|
||||
project_id: The project ID
|
||||
"""
|
||||
client = _get_basecamp_client()
|
||||
if not client:
|
||||
return _get_auth_error_response()
|
||||
|
||||
try:
|
||||
comments = await _run_sync(client.get_comments, project_id, recording_id)
|
||||
return {
|
||||
"status": "success",
|
||||
"comments": comments,
|
||||
"count": len(comments)
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting comments: {e}")
|
||||
if "401" in str(e) and "expired" in str(e).lower():
|
||||
return {
|
||||
"error": "OAuth token expired",
|
||||
"message": "Your Basecamp OAuth token expired during the API call. Please re-authenticate by visiting http://localhost:8000 and completing the OAuth flow again."
|
||||
}
|
||||
return {
|
||||
"error": "Execution error",
|
||||
"message": str(e)
|
||||
}
|
||||
|
||||
@mcp.tool()
|
||||
async def get_campfire_lines(project_id: str, campfire_id: str) -> Dict[str, Any]:
|
||||
"""Get recent messages from a Basecamp campfire (chat room).
|
||||
|
||||
Args:
|
||||
project_id: The project ID
|
||||
campfire_id: The campfire/chat room ID
|
||||
"""
|
||||
client = _get_basecamp_client()
|
||||
if not client:
|
||||
return _get_auth_error_response()
|
||||
|
||||
try:
|
||||
lines = await _run_sync(client.get_campfire_lines, project_id, campfire_id)
|
||||
return {
|
||||
"status": "success",
|
||||
"campfire_lines": lines,
|
||||
"count": len(lines)
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting campfire lines: {e}")
|
||||
if "401" in str(e) and "expired" in str(e).lower():
|
||||
return {
|
||||
"error": "OAuth token expired",
|
||||
"message": "Your Basecamp OAuth token expired during the API call. Please re-authenticate by visiting http://localhost:8000 and completing the OAuth flow again."
|
||||
}
|
||||
return {
|
||||
"error": "Execution error",
|
||||
"message": str(e)
|
||||
}
|
||||
|
||||
@mcp.tool()
|
||||
async def get_card_tables(project_id: str) -> Dict[str, Any]:
|
||||
"""Get all card tables for a project.
|
||||
|
||||
Args:
|
||||
project_id: The project ID
|
||||
"""
|
||||
client = _get_basecamp_client()
|
||||
if not client:
|
||||
return _get_auth_error_response()
|
||||
|
||||
try:
|
||||
card_tables = await _run_sync(client.get_card_tables, project_id)
|
||||
return {
|
||||
"status": "success",
|
||||
"card_tables": card_tables,
|
||||
"count": len(card_tables)
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting card tables: {e}")
|
||||
if "401" in str(e) and "expired" in str(e).lower():
|
||||
return {
|
||||
"error": "OAuth token expired",
|
||||
"message": "Your Basecamp OAuth token expired during the API call. Please re-authenticate by visiting http://localhost:8000 and completing the OAuth flow again."
|
||||
}
|
||||
return {
|
||||
"error": "Execution error",
|
||||
"message": str(e)
|
||||
}
|
||||
|
||||
@mcp.tool()
|
||||
async def get_card_table(project_id: str) -> Dict[str, Any]:
|
||||
"""Get the card table details for a project.
|
||||
|
||||
Args:
|
||||
project_id: The project ID
|
||||
"""
|
||||
client = _get_basecamp_client()
|
||||
if not client:
|
||||
return _get_auth_error_response()
|
||||
|
||||
try:
|
||||
card_table = await _run_sync(client.get_card_table, project_id)
|
||||
card_table_details = await _run_sync(client.get_card_table_details, project_id, card_table['id'])
|
||||
return {
|
||||
"status": "success",
|
||||
"card_table": card_table_details
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting card table: {e}")
|
||||
error_msg = str(e)
|
||||
if "401" in str(e) and "expired" in str(e).lower():
|
||||
return {
|
||||
"error": "OAuth token expired",
|
||||
"message": "Your Basecamp OAuth token expired during the API call. Please re-authenticate by visiting http://localhost:8000 and completing the OAuth flow again."
|
||||
}
|
||||
return {
|
||||
"status": "error",
|
||||
"message": f"Error getting card table: {error_msg}",
|
||||
"debug": error_msg
|
||||
}
|
||||
|
||||
@mcp.tool()
|
||||
async def get_columns(project_id: str, card_table_id: str) -> Dict[str, Any]:
|
||||
"""Get all columns in a card table.
|
||||
|
||||
Args:
|
||||
project_id: The project ID
|
||||
card_table_id: The card table ID
|
||||
"""
|
||||
client = _get_basecamp_client()
|
||||
if not client:
|
||||
return _get_auth_error_response()
|
||||
|
||||
try:
|
||||
columns = await _run_sync(client.get_columns, project_id, card_table_id)
|
||||
return {
|
||||
"status": "success",
|
||||
"columns": columns,
|
||||
"count": len(columns)
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting columns: {e}")
|
||||
if "401" in str(e) and "expired" in str(e).lower():
|
||||
return {
|
||||
"error": "OAuth token expired",
|
||||
"message": "Your Basecamp OAuth token expired during the API call. Please re-authenticate by visiting http://localhost:8000 and completing the OAuth flow again."
|
||||
}
|
||||
return {
|
||||
"error": "Execution error",
|
||||
"message": str(e)
|
||||
}
|
||||
|
||||
@mcp.tool()
|
||||
async def get_cards(project_id: str, column_id: str) -> Dict[str, Any]:
|
||||
"""Get all cards in a column.
|
||||
|
||||
Args:
|
||||
project_id: The project ID
|
||||
column_id: The column ID
|
||||
"""
|
||||
client = _get_basecamp_client()
|
||||
if not client:
|
||||
return _get_auth_error_response()
|
||||
|
||||
try:
|
||||
cards = await _run_sync(client.get_cards, project_id, column_id)
|
||||
return {
|
||||
"status": "success",
|
||||
"cards": cards,
|
||||
"count": len(cards)
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting cards: {e}")
|
||||
if "401" in str(e) and "expired" in str(e).lower():
|
||||
return {
|
||||
"error": "OAuth token expired",
|
||||
"message": "Your Basecamp OAuth token expired during the API call. Please re-authenticate by visiting http://localhost:8000 and completing the OAuth flow again."
|
||||
}
|
||||
return {
|
||||
"error": "Execution error",
|
||||
"message": str(e)
|
||||
}
|
||||
|
||||
@mcp.tool()
|
||||
async def create_card(project_id: str, column_id: str, title: str, content: Optional[str] = None, due_on: Optional[str] = None, notify: bool = False) -> Dict[str, Any]:
|
||||
"""Create a new card in a column.
|
||||
|
||||
Args:
|
||||
project_id: The project ID
|
||||
column_id: The column ID
|
||||
title: The card title
|
||||
content: Optional card content/description
|
||||
due_on: Optional due date (ISO 8601 format)
|
||||
notify: Whether to notify assignees (default: false)
|
||||
"""
|
||||
client = _get_basecamp_client()
|
||||
if not client:
|
||||
return _get_auth_error_response()
|
||||
|
||||
try:
|
||||
card = await _run_sync(client.create_card, project_id, column_id, title, content, due_on, notify)
|
||||
return {
|
||||
"status": "success",
|
||||
"card": card,
|
||||
"message": f"Card '{title}' created successfully"
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"Error creating card: {e}")
|
||||
if "401" in str(e) and "expired" in str(e).lower():
|
||||
return {
|
||||
"error": "OAuth token expired",
|
||||
"message": "Your Basecamp OAuth token expired during the API call. Please re-authenticate by visiting http://localhost:8000 and completing the OAuth flow again."
|
||||
}
|
||||
return {
|
||||
"error": "Execution error",
|
||||
"message": str(e)
|
||||
}
|
||||
|
||||
@mcp.tool()
|
||||
async def get_column(project_id: str, column_id: str) -> Dict[str, Any]:
|
||||
"""Get details for a specific column.
|
||||
|
||||
Args:
|
||||
project_id: The project ID
|
||||
column_id: The column ID
|
||||
"""
|
||||
client = _get_basecamp_client()
|
||||
if not client:
|
||||
return _get_auth_error_response()
|
||||
|
||||
try:
|
||||
column = await _run_sync(client.get_column, project_id, column_id)
|
||||
return {
|
||||
"status": "success",
|
||||
"column": column
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting column: {e}")
|
||||
if "401" in str(e) and "expired" in str(e).lower():
|
||||
return {
|
||||
"error": "OAuth token expired",
|
||||
"message": "Your Basecamp OAuth token expired during the API call. Please re-authenticate by visiting http://localhost:8000 and completing the OAuth flow again."
|
||||
}
|
||||
return {
|
||||
"error": "Execution error",
|
||||
"message": str(e)
|
||||
}
|
||||
|
||||
@mcp.tool()
|
||||
async def create_column(project_id: str, card_table_id: str, title: str) -> Dict[str, Any]:
|
||||
"""Create a new column in a card table.
|
||||
|
||||
Args:
|
||||
project_id: The project ID
|
||||
card_table_id: The card table ID
|
||||
title: The column title
|
||||
"""
|
||||
client = _get_basecamp_client()
|
||||
if not client:
|
||||
return _get_auth_error_response()
|
||||
|
||||
try:
|
||||
column = await _run_sync(client.create_column, project_id, card_table_id, title)
|
||||
return {
|
||||
"status": "success",
|
||||
"column": column,
|
||||
"message": f"Column '{title}' created successfully"
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"Error creating column: {e}")
|
||||
if "401" in str(e) and "expired" in str(e).lower():
|
||||
return {
|
||||
"error": "OAuth token expired",
|
||||
"message": "Your Basecamp OAuth token expired during the API call. Please re-authenticate by visiting http://localhost:8000 and completing the OAuth flow again."
|
||||
}
|
||||
return {
|
||||
"error": "Execution error",
|
||||
"message": str(e)
|
||||
}
|
||||
|
||||
@mcp.tool()
|
||||
async def move_card(project_id: str, card_id: str, column_id: str) -> Dict[str, Any]:
|
||||
"""Move a card to a new column.
|
||||
|
||||
Args:
|
||||
project_id: The project ID
|
||||
card_id: The card ID
|
||||
column_id: The destination column ID
|
||||
"""
|
||||
client = _get_basecamp_client()
|
||||
if not client:
|
||||
return _get_auth_error_response()
|
||||
|
||||
try:
|
||||
await _run_sync(client.move_card, project_id, card_id, column_id)
|
||||
return {
|
||||
"status": "success",
|
||||
"message": f"Card moved to column {column_id}"
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"Error moving card: {e}")
|
||||
if "401" in str(e) and "expired" in str(e).lower():
|
||||
return {
|
||||
"error": "OAuth token expired",
|
||||
"message": "Your Basecamp OAuth token expired during the API call. Please re-authenticate by visiting http://localhost:8000 and completing the OAuth flow again."
|
||||
}
|
||||
return {
|
||||
"error": "Execution error",
|
||||
"message": str(e)
|
||||
}
|
||||
|
||||
@mcp.tool()
|
||||
async def complete_card(project_id: str, card_id: str) -> Dict[str, Any]:
|
||||
"""Mark a card as complete.
|
||||
|
||||
Args:
|
||||
project_id: The project ID
|
||||
card_id: The card ID
|
||||
"""
|
||||
client = _get_basecamp_client()
|
||||
if not client:
|
||||
return _get_auth_error_response()
|
||||
|
||||
try:
|
||||
await _run_sync(client.complete_card, project_id, card_id)
|
||||
return {
|
||||
"status": "success",
|
||||
"message": "Card marked as complete"
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"Error completing card: {e}")
|
||||
if "401" in str(e) and "expired" in str(e).lower():
|
||||
return {
|
||||
"error": "OAuth token expired",
|
||||
"message": "Your Basecamp OAuth token expired during the API call. Please re-authenticate by visiting http://localhost:8000 and completing the OAuth flow again."
|
||||
}
|
||||
return {
|
||||
"error": "Execution error",
|
||||
"message": str(e)
|
||||
}
|
||||
|
||||
@mcp.tool()
|
||||
async def get_card(project_id: str, card_id: str) -> Dict[str, Any]:
|
||||
"""Get details for a specific card.
|
||||
|
||||
Args:
|
||||
project_id: The project ID
|
||||
card_id: The card ID
|
||||
"""
|
||||
client = _get_basecamp_client()
|
||||
if not client:
|
||||
return _get_auth_error_response()
|
||||
|
||||
try:
|
||||
card = await _run_sync(client.get_card, project_id, card_id)
|
||||
return {
|
||||
"status": "success",
|
||||
"card": card
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting card: {e}")
|
||||
if "401" in str(e) and "expired" in str(e).lower():
|
||||
return {
|
||||
"error": "OAuth token expired",
|
||||
"message": "Your Basecamp OAuth token expired during the API call. Please re-authenticate by visiting http://localhost:8000 and completing the OAuth flow again."
|
||||
}
|
||||
return {
|
||||
"error": "Execution error",
|
||||
"message": str(e)
|
||||
}
|
||||
|
||||
@mcp.tool()
|
||||
async def update_card(project_id: str, card_id: str, title: Optional[str] = None, content: Optional[str] = None, due_on: Optional[str] = None, assignee_ids: Optional[List[str]] = None) -> Dict[str, Any]:
|
||||
"""Update a card.
|
||||
|
||||
Args:
|
||||
project_id: The project ID
|
||||
card_id: The card ID
|
||||
title: The new card title
|
||||
content: The new card content/description
|
||||
due_on: Due date (ISO 8601 format)
|
||||
assignee_ids: Array of person IDs to assign to the card
|
||||
"""
|
||||
client = _get_basecamp_client()
|
||||
if not client:
|
||||
return _get_auth_error_response()
|
||||
|
||||
try:
|
||||
card = await _run_sync(client.update_card, project_id, card_id, title, content, due_on, assignee_ids)
|
||||
return {
|
||||
"status": "success",
|
||||
"card": card,
|
||||
"message": "Card updated successfully"
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"Error updating card: {e}")
|
||||
if "401" in str(e) and "expired" in str(e).lower():
|
||||
return {
|
||||
"error": "OAuth token expired",
|
||||
"message": "Your Basecamp OAuth token expired during the API call. Please re-authenticate by visiting http://localhost:8000 and completing the OAuth flow again."
|
||||
}
|
||||
return {
|
||||
"error": "Execution error",
|
||||
"message": str(e)
|
||||
}
|
||||
|
||||
# Core FastMCP server with essential Basecamp functionality
|
||||
# Additional tools can be added incrementally as needed
|
||||
|
||||
if __name__ == "__main__":
|
||||
logger.info("Starting Basecamp FastMCP server")
|
||||
# Run using official MCP stdio transport
|
||||
mcp.run(transport='stdio')
|
||||
@@ -3,3 +3,6 @@ python-dotenv==1.0.0
|
||||
flask==2.3.3
|
||||
flask-cors==4.0.0
|
||||
pytest==7.4.0
|
||||
mcp[cli]>=1.2.0
|
||||
httpx>=0.25.0
|
||||
anyio>=4.0.0
|
||||
|
||||
Reference in New Issue
Block a user