m3tam3re/basecamp-mcp-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Fix token_storage function name and ensure Composio integration is fully functional

Browse Source
This commit is contained in:
George Antonopoulos
2025-03-09 17:29:28 +00:00
parent e08e48de50
commit 712344741e
5 changed files with 775 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

79
README.md
View File

@@ -49,6 +49,7 @@ The project consists of the following components:
BASECAMP_ACCOUNT_ID=your_account_id
FLASK_SECRET_KEY=random_secret_key
MCP_API_KEY=your_api_key
COMPOSIO_API_KEY=your_composio_api_key
```
## Usage
@@ -163,3 +164,81 @@ This project is licensed under the MIT License - see the LICENSE file for detail
- Fixed CORS support by adding the Flask-CORS package
These changes ensure more reliable and consistent responses from the MCP server, particularly for Campfire chat functionality.
### March 9, 2024 - Added Composio Integration
Added support for [Composio](https://composio.dev/) integration, allowing the Basecamp MCP server to be used with Composio for AI-powered workflows. This integration follows the Model Context Protocol (MCP) standards and includes:
- New endpoints for Composio compatibility:
- `/composio/schema` - Returns the schema of available tools in Composio-compatible format
- `/composio/tool` - Handles Composio tool calls with standardized parameters
- `/composio/check_auth` - Checks authentication status for Composio requests
- Standardized tool naming and parameter formats to work with Composio's MCP specifications
- A standalone example client for testing and demonstrating the integration
## Using with Composio
### Prerequisites
1. Create a Composio account at [https://app.composio.dev](https://app.composio.dev)
2. Obtain a Composio API key from your Composio dashboard
3. Add your API key to your `.env` file:
```
COMPOSIO_API_KEY=your_composio_api_key
```
### Setting Up Composio Integration
1. Make sure you have authenticated with Basecamp using the OAuth app (http://localhost:8000/)
2. Run the MCP server with the Composio integration enabled:
```
python mcp_server.py
```
3. In your Composio dashboard, add a new custom integration:
- Integration URL: `http://localhost:5001/composio/schema`
- Authentication: OAuth (managed by our implementation)
4. You can now use Composio to connect to your Basecamp account through the MCP server:
- Composio will discover available tools via the schema endpoint
- Tool executions will be handled by the `/composio/tool` endpoint
- Authentication status is checked via the `/composio/check_auth` endpoint
### Example Composio Client
We provide a simple client example in `composio_client_example.py` that demonstrates how to:
1. Check authentication status
2. Retrieve the tool schema
3. Execute various Basecamp operations through the Composio integration
Run the example with:
```
python composio_client_example.py
```
### Testing the Integration
To test the integration without connecting to Composio:
1. Run the MCP server:
```
python mcp_server.py
```
2. Use curl to test the endpoints directly:
```bash
# Check authentication status
curl http://localhost:5001/composio/check_auth
# Get the schema
curl http://localhost:5001/composio/schema
# Execute a tool (get projects)
curl -X POST http://localhost:5001/composio/tool \
-H "Content-Type: application/json" \
-d '{"tool": "GET_PROJECTS", "params": {}}'
```
For more detailed documentation on Composio integration, refer to the [official Composio documentation](https://docs.composio.dev).

185
composio_client_example.py Normal file
View File

@@ -0,0 +1,185 @@
#!/usr/bin/env python
"""
Example client for using the Basecamp MCP server with Composio.
This example demonstrates MCP protocol integration requirements.
"""
import os
import json
import requests
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
# Configuration
MCP_SERVER_URL = "http://localhost:5001"
COMPOSIO_API_KEY = os.getenv("COMPOSIO_API_KEY")
class BasecampComposioClient:
"""A client for interacting with Basecamp through Composio's MCP protocol."""
def __init__(self, mcp_server_url=MCP_SERVER_URL):
"""Initialize the client with the MCP server URL."""
self.mcp_server_url = mcp_server_url
self.headers = {
"Content-Type": "application/json"
}
# Add Composio API key if available
if COMPOSIO_API_KEY:
self.headers["X-Composio-API-Key"] = COMPOSIO_API_KEY
def check_auth(self):
"""Check if OAuth authentication is available."""
response = requests.get(
f"{self.mcp_server_url}/composio/check_auth",
headers=self.headers
)
return response.json()
def get_schema(self):
"""Get the schema of available tools."""
response = requests.get(
f"{self.mcp_server_url}/composio/schema",
headers=self.headers
)
return response.json()
def execute_tool(self, tool_name, params=None):
"""Execute a tool with the given parameters."""
if params is None:
params = {}
response = requests.post(
f"{self.mcp_server_url}/composio/tool",
headers=self.headers,
json={"tool": tool_name, "params": params}
)
return response.json()
def get_projects(self):
"""Get all projects from Basecamp."""
return self.execute_tool("GET_PROJECTS")
def get_project(self, project_id):
"""Get details for a specific project."""
return self.execute_tool("GET_PROJECT", {"project_id": project_id})
def get_todolists(self, project_id):
"""Get all todo lists for a project."""
return self.execute_tool("GET_TODOLISTS", {"project_id": project_id})
def get_todos(self, todolist_id):
"""Get all todos for a specific todolist."""
return self.execute_tool("GET_TODOS", {"todolist_id": todolist_id})
def get_campfire(self, project_id):
"""Get all chat rooms (campfires) for a project."""
return self.execute_tool("GET_CAMPFIRE", {"project_id": project_id})
def get_campfire_lines(self, project_id, campfire_id):
"""Get messages from a specific chat room."""
return self.execute_tool("GET_CAMPFIRE_LINES", {
"project_id": project_id,
"campfire_id": campfire_id
})
def search(self, query, project_id=None):
"""Search across Basecamp resources."""
params = {"query": query}
if project_id:
params["project_id"] = project_id
return self.execute_tool("SEARCH", params)
def get_comments(self, recording_id, bucket_id):
"""Get comments for a specific Basecamp object."""
return self.execute_tool("GET_COMMENTS", {
"recording_id": recording_id,
"bucket_id": bucket_id
})
def create_comment(self, recording_id, bucket_id, content):
"""Create a new comment on a Basecamp object."""
return self.execute_tool("CREATE_COMMENT", {
"recording_id": recording_id,
"bucket_id": bucket_id,
"content": content
})
def main():
"""Main function to demonstrate the client."""
client = BasecampComposioClient()
# Verify connectivity to MCP server
print("==== Basecamp MCP-Composio Integration Test ====")
# Check authentication
print("\n1. Checking authentication status...")
auth_status = client.check_auth()
print(f"Authentication Status: {json.dumps(auth_status, indent=2)}")
if auth_status.get("status") == "error":
print(f"Please authenticate at: {auth_status.get('error', {}).get('auth_url', '')}")
return
# Get available tools
print("\n2. Retrieving tool schema...")
schema = client.get_schema()
print(f"Server Name: {schema.get('name')}")
print(f"Version: {schema.get('version')}")
print(f"Authentication Type: {schema.get('auth', {}).get('type')}")
print("\nAvailable Tools:")
for tool in schema.get("tools", []):
required_params = tool.get("parameters", {}).get("required", [])
required_str = ", ".join(required_params) if required_params else "None"
print(f"- {tool['name']}: {tool['description']} (Required params: {required_str})")
# Get projects
print("\n3. Fetching projects...")
projects_response = client.get_projects()
if projects_response.get("status") == "error":
print(f"Error: {projects_response.get('error', {}).get('message', 'Unknown error')}")
else:
project_data = projects_response.get("data", [])
print(f"Found {len(project_data)} projects:")
for i, project in enumerate(project_data[:3], 1): # Show first 3 projects
print(f"{i}. {project.get('name')} (ID: {project.get('id')})")
if project_data:
# Use the first project for further examples
project_id = project_data[0].get("id")
# Get campfires for the project
print(f"\n4. Fetching campfires for project {project_id}...")
campfires_response = client.get_campfire(project_id)
if campfires_response.get("status") == "error":
print(f"Error: {campfires_response.get('error', {}).get('message', 'Unknown error')}")
else:
campfire_data = campfires_response.get("data", {}).get("campfire", [])
print(f"Found {len(campfire_data)} campfires:")
for i, campfire in enumerate(campfire_data[:2], 1): # Show first 2 campfires
print(f"{i}. {campfire.get('title')} (ID: {campfire.get('id')})")
if campfire_data:
# Get messages from the first campfire
campfire_id = campfire_data[0].get("id")
print(f"\n5. Fetching messages from campfire {campfire_id}...")
messages_response = client.get_campfire_lines(project_id, campfire_id)
if messages_response.get("status") == "error":
print(f"Error: {messages_response.get('error', {}).get('message', 'Unknown error')}")
else:
message_data = messages_response.get("data", {}).get("lines", [])
print(f"Found {len(message_data)} messages:")
for i, message in enumerate(message_data[:3], 1): # Show first 3 messages
creator = message.get("creator", {}).get("name", "Unknown")
content = message.get("title", "No content")
print(f"{i}. From {creator}: {content[:50]}...")
print("\n==== Test completed ====")
print("This example demonstrates how to connect to the Basecamp MCP server")
print("and use it with the Composio MCP protocol.")
if __name__ == "__main__":
main()

420
composio_integration.py Normal file
View File

@@ -0,0 +1,420 @@
#!/usr/bin/env python
import os
import json
import logging
from flask import Flask, request, jsonify
from basecamp_client import BasecampClient
from search_utils import BasecampSearch
import token_storage
from dotenv import load_dotenv
# Set up logging
logging.basicConfig(level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# Load environment variables
load_dotenv()
def get_basecamp_client(auth_mode='oauth'):
"""
Returns a BasecampClient instance with appropriate authentication.
Args:
auth_mode (str): The authentication mode to use ('oauth' or 'basic')
Returns:
BasecampClient: A client for interacting with Basecamp
"""
if auth_mode == 'oauth':
# Get the OAuth token
token_data = token_storage.get_token()
if not token_data or 'access_token' not in token_data:
logger.error("No OAuth token available")
return None
# Create a client using the OAuth token
account_id = os.getenv('BASECAMP_ACCOUNT_ID')
user_agent = os.getenv('USER_AGENT')
client = BasecampClient(
account_id=account_id,
user_agent=user_agent,
access_token=token_data['access_token'],
auth_mode='oauth'
)
return client
else:
# Basic auth is not recommended but keeping for compatibility
username = os.getenv('BASECAMP_USERNAME')
password = os.getenv('BASECAMP_PASSWORD')
account_id = os.getenv('BASECAMP_ACCOUNT_ID')
user_agent = os.getenv('USER_AGENT')
client = BasecampClient(
username=username,
password=password,
account_id=account_id,
user_agent=user_agent,
auth_mode='basic'
)
return client
def get_schema():
"""
Returns the schema for Basecamp tools compatible with Composio's MCP format.
Returns:
dict: A schema describing available tools and their parameters according to Composio specs
"""
schema = {
"name": "Basecamp MCP Server",
"description": "Integration with Basecamp 3 for project management and team collaboration",
"version": "1.0.0",
"auth": {
"type": "oauth2",
"redirect_url": "http://localhost:8000",
"token_url": "http://localhost:8000/token/info"
},
"contact": {
"name": "Basecamp MCP Server Team",
"url": "https://github.com/georgeantonopoulos/Basecamp-MCP-Server"
},
"tools": [
{
"name": "GET_PROJECTS",
"description": "Get all projects from Basecamp",
"parameters": {
"type": "object",
"properties": {},
"required": []
}
},
{
"name": "GET_PROJECT",
"description": "Get details for a specific project",
"parameters": {
"type": "object",
"properties": {
"project_id": {"type": "string", "description": "The ID of the project"}
},
"required": ["project_id"]
}
},
{
"name": "GET_TODOLISTS",
"description": "Get all todo lists for a project",
"parameters": {
"type": "object",
"properties": {
"project_id": {"type": "string", "description": "The ID of the project"}
},
"required": ["project_id"]
}
},
{
"name": "GET_TODOS",
"description": "Get all todos for a specific todolist",
"parameters": {
"type": "object",
"properties": {
"todolist_id": {"type": "string", "description": "The ID of the todolist"}
},
"required": ["todolist_id"]
}
},
{
"name": "GET_CAMPFIRE",
"description": "Get all chat rooms (campfires) for a project",
"parameters": {
"type": "object",
"properties": {
"project_id": {"type": "string", "description": "The ID of the project"}
},
"required": ["project_id"]
}
},
{
"name": "GET_CAMPFIRE_LINES",
"description": "Get messages from a specific chat room",
"parameters": {
"type": "object",
"properties": {
"project_id": {"type": "string", "description": "The ID of the project"},
"campfire_id": {"type": "string", "description": "The ID of the campfire/chat room"}
},
"required": ["project_id", "campfire_id"]
}
},
{
"name": "SEARCH",
"description": "Search across Basecamp resources",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "The search query"},
"project_id": {"type": "string", "description": "Optional project ID to limit search scope"}
},
"required": ["query"]
}
},
{
"name": "GET_COMMENTS",
"description": "Get comments for a specific Basecamp object",
"parameters": {
"type": "object",
"properties": {
"recording_id": {"type": "string", "description": "The ID of the object to get comments for"},
"bucket_id": {"type": "string", "description": "The bucket ID"}
},
"required": ["recording_id", "bucket_id"]
}
},
{
"name": "CREATE_COMMENT",
"description": "Create a new comment on a Basecamp object",
"parameters": {
"type": "object",
"properties": {
"recording_id": {"type": "string", "description": "The ID of the object to comment on"},
"bucket_id": {"type": "string", "description": "The bucket ID"},
"content": {"type": "string", "description": "The comment content"}
},
"required": ["recording_id", "bucket_id", "content"]
}
}
]
}
return schema
def handle_composio_request(data):
"""
Handle a request from Composio following MCP standards.
Args:
data (dict): The request data containing tool name and parameters
Returns:
dict: The result of the tool execution in MCP-compliant format
"""
# Check if the API key is valid (if provided)
composio_api_key = os.getenv('COMPOSIO_API_KEY')
request_api_key = request.headers.get('X-Composio-API-Key')
if composio_api_key and request_api_key and composio_api_key != request_api_key:
return {
"status": "error",
"error": {
"type": "authentication_error",
"message": "Invalid API key provided"
}
}
tool_name = data.get('tool')
params = data.get('params', {})
# Get a Basecamp client
client = get_basecamp_client(auth_mode='oauth')
if not client:
return {
"status": "error",
"error": {
"type": "authentication_required",
"message": "OAuth authentication required",
"auth_url": "http://localhost:8000/"
}
}
# Route to the appropriate handler based on tool_name
try:
if tool_name == "GET_PROJECTS":
result = client.get_projects()
return {
"status": "success",
"data": result
}
elif tool_name == "GET_PROJECT":
project_id = params.get('project_id')
if not project_id:
return {
"status": "error",
"error": {
"type": "invalid_parameters",
"message": "Missing required parameter: project_id"
}
}
result = client.get_project(project_id)
return {
"status": "success",
"data": result
}
elif tool_name == "GET_TODOLISTS":
project_id = params.get('project_id')
if not project_id:
return {
"status": "error",
"error": {
"type": "invalid_parameters",
"message": "Missing required parameter: project_id"
}
}
result = client.get_todolists(project_id)
return {
"status": "success",
"data": result
}
elif tool_name == "GET_TODOS":
todolist_id = params.get('todolist_id')
if not todolist_id:
return {
"status": "error",
"error": {
"type": "invalid_parameters",
"message": "Missing required parameter: todolist_id"
}
}
result = client.get_todos(todolist_id)
return {
"status": "success",
"data": result
}
elif tool_name == "GET_CAMPFIRE":
project_id = params.get('project_id')
if not project_id:
return {
"status": "error",
"error": {
"type": "invalid_parameters",
"message": "Missing required parameter: project_id"
}
}
result = client.get_campfires(project_id)
return {
"status": "success",
"data": {
"campfire": result
}
}
elif tool_name == "GET_CAMPFIRE_LINES":
project_id = params.get('project_id')
campfire_id = params.get('campfire_id')
if not project_id or not campfire_id:
return {
"status": "error",
"error": {
"type": "invalid_parameters",
"message": "Missing required parameters: project_id and/or campfire_id"
}
}
result = client.get_campfire_lines(project_id, campfire_id)
return {
"status": "success",
"data": result
}
elif tool_name == "SEARCH":
query = params.get('query')
project_id = params.get('project_id')
if not query:
return {
"status": "error",
"error": {
"type": "invalid_parameters",
"message": "Missing required parameter: query"
}
}
search = BasecampSearch(client=client)
results = []
# Search projects
if not project_id:
projects = search.search_projects(query)
if projects:
results.extend([{"type": "project", "data": p} for p in projects])
# If project_id is provided, search within that project
if project_id:
# Search todolists
todolists = search.search_todolists(query, project_id)
if todolists:
results.extend([{"type": "todolist", "data": t} for t in todolists])
# Search todos
todos = search.search_todos(query, project_id)
if todos:
results.extend([{"type": "todo", "data": t} for t in todos])
# Search campfire lines
campfires = client.get_campfires(project_id)
for campfire in campfires:
campfire_id = campfire.get('id')
lines = search.search_campfire_lines(query, project_id, campfire_id)
if lines:
results.extend([{"type": "campfire_line", "data": l} for l in lines])
return {
"status": "success",
"data": {
"results": results,
"count": len(results)
}
}
elif tool_name == "GET_COMMENTS":
recording_id = params.get('recording_id')
bucket_id = params.get('bucket_id')
if not recording_id or not bucket_id:
return {
"status": "error",
"error": {
"type": "invalid_parameters",
"message": "Missing required parameters: recording_id and/or bucket_id"
}
}
result = client.get_comments(recording_id, bucket_id)
return {
"status": "success",
"data": result
}
elif tool_name == "CREATE_COMMENT":
recording_id = params.get('recording_id')
bucket_id = params.get('bucket_id')
content = params.get('content')
if not recording_id or not bucket_id or not content:
return {
"status": "error",
"error": {
"type": "invalid_parameters",
"message": "Missing required parameters"
}
}
result = client.create_comment(recording_id, bucket_id, content)
return {
"status": "success",
"data": result
}
else:
return {
"status": "error",
"error": {
"type": "unknown_tool",
"message": f"Unknown tool: {tool_name}"
}
}
except Exception as e:
logger.error(f"Error handling tool {tool_name}: {str(e)}")
return {
"status": "error",
"error": {
"type": "server_error",
"message": f"Error executing tool: {str(e)}"
}
}

147
mcp_server.py
View File

@@ -34,6 +34,14 @@ except Exception as e:
traceback.print_exc()
sys.exit(1)
# Import Composio integration components
try:
from composio_integration import get_schema, handle_composio_request
except Exception as e:
print(f"Error importing Composio integration: {str(e)}")
traceback.print_exc()
# Don't exit since Composio integration is optional
# Helper function for consistent response format
def mcp_response(data=None, status="success", error=None, message=None, status_code=200):
"""
@@ -243,65 +251,38 @@ def handle_options(path):
# MCP Info endpoint
@app.route('/mcp/info', methods=['GET'])
def mcp_info():
"""Return information about this MCP server."""
logger.info("MCP info endpoint called")
try:
# Keep this operation lightweight - no external API calls here
return jsonify({
"name": "Basecamp",
"version": "1.0.0",
"description": "Basecamp 3 API integration for Cursor",
"author": "Cursor",
"actions": [
{
"name": "get_required_parameters",
"description": "Get required parameters for connecting to Basecamp"
},
{
"name": "initiate_connection",
"description": "Connect to Basecamp using credentials"
},
{
"name": "check_active_connection",
"description": "Check if the connection to Basecamp is active"
},
{
"name": "get_projects",
"description": "Get all projects with optional filtering"
},
{
"name": "get_todo_lists",
"description": "Get all to-do lists for a project"
},
{
"name": "get_todos",
"description": "Get all to-dos with various filters"
},
{
"name": "get_comments",
"description": "Get comments for a specific recording (todo, message, etc.)"
},
{
"name": "create_comment",
"description": "Create a comment on a recording"
},
{
"name": "update_comment",
"description": "Update a comment"
},
{
"name": "delete_comment",
"description": "Delete a comment"
},
{
"name": "search_all",
"description": "Search across all Basecamp resources"
}
]
})
except Exception as e:
logger.error(f"Error in mcp_info: {str(e)}", exc_info=True)
return jsonify({"status": "error", "message": str(e)}), 500
"""
Provides information about this MCP server.
"""
server_info = {
"name": "Basecamp MCP Server",
"version": "1.0.0",
"description": "A MCP server for Basecamp 3",
"auth_modes": ["oauth"],
"actions": [
"get_projects",
"get_project",
"get_todosets",
"get_todolists",
"get_todolist",
"get_todos",
"get_todo",
"get_people",
"get_campfire",
"get_campfire_lines",
"get_message_board",
"get_messages",
"get_schedule",
"get_schedule_entries",
"get_comments",
"create_comment",
"update_comment",
"delete_comment",
"search"
],
"composio_compatible": True # Add this flag to indicate Composio compatibility
}
return jsonify(server_info)
# MCP Action endpoint with improved error handling
@app.route('/mcp/action', methods=['POST'])
@@ -783,6 +764,54 @@ def tool(connection_id):
"error": str(e)
}), 500
# Add Composio-specific routes
@app.route('/composio/schema', methods=['GET'])
def composio_schema():
"""
Returns the schema for Basecamp tools compatible with Composio.
"""
try:
schema = get_schema()
return jsonify(schema)
except Exception as e:
logger.error(f"Error generating Composio schema: {str(e)}")
return jsonify({"error": "server_error", "message": f"Error generating schema: {str(e)}"}), 500
@app.route('/composio/tool', methods=['POST'])
def composio_tool():
"""
Handles tool calls from Composio.
"""
try:
data = request.json
if not data:
return jsonify({"error": "invalid_request", "message": "Invalid request data"}), 400
result = handle_composio_request(data)
return jsonify(result)
except Exception as e:
logger.error(f"Error handling Composio tool call: {str(e)}")
return jsonify({"error": "server_error", "message": f"Error handling tool call: {str(e)}"}), 500
@app.route('/composio/check_auth', methods=['GET'])
def composio_check_auth():
"""
Checks if OAuth authentication is available for Composio.
"""
try:
token_data = token_storage.get_token()
if token_data and 'access_token' in token_data:
return jsonify({"authenticated": True})
else:
return jsonify({
"authenticated": False,
"auth_url": f"http://localhost:8000/",
"message": "OAuth authentication required. Please visit the auth URL to authenticate."
})
except Exception as e:
logger.error(f"Error checking Composio auth: {str(e)}")
return jsonify({"error": "server_error", "message": f"Error checking auth: {str(e)}"}), 500
if __name__ == '__main__':
try:
logger.info(f"Starting MCP server on port {MCP_PORT}")

1
requirements.txt
View File

@@ -1,3 +1,4 @@
requests==2.31.0
python-dotenv==1.0.0
flask==2.3.3
flask-cors==4.0.0