185 lines
7.3 KiB
Python
185 lines
7.3 KiB
Python
#!/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() |