HubSpot MCP Server

import logging import json import os from dotenv import load_dotenv load_dotenv() from datetime import datetime from typing import Any, Dict, List logger = logging.getLogger('mcp_shared_space') # In-memory storage for demonstration. # In production, replace this with a persistent datastore. SHARED_SUMMARIES: Dict[str, Dict[str, Any]] = {} def load_user_roles(company_id: str) -> Dict[str, str]: """ Dynamically load user roles for a given company. Replace this with a DB query or external API in production. Expected structure (in a JSON file, e.g., user_roles.json): { "think41": { "admin@think41.com": "admin", "manager1@think41.com": "manager", "employee1@think41.com": "employee" }, "xpay": { "admin@xpay.com": "admin", "manager@xpay.com": "manager" } } """ roles_file = os.getenv("USER_ROLES_FILE", "user_roles.json") if os.path.exists(roles_file): with open(roles_file, "r") as f: all_roles = json.load(f) return all_roles.get(company_id, {}) else: # If file not found, return empty dict. return {} def get_user_role(user_id: str, company_id: str) -> str: """ Retrieve the role of a user dynamically based on the company. Defaults to 'employee' if the user is not found. """ roles = load_user_roles(company_id) return roles.get(user_id, "employee") def generate_summary_id() -> str: """Generate a unique summary ID based on the current UTC timestamp.""" return f"summary_{int(datetime.utcnow().timestamp() * 1000)}" def create_summary(user_id: str, company_id: str, summary_text: str) -> Dict[str, Any]: """ Create a new conversation summary in the shared space. Only 'manager' or 'admin' roles are authorized. """ role = get_user_role(user_id, company_id) if role not in ("manager", "admin"): raise PermissionError("User not authorized to create a summary.") summary_id = generate_summary_id() summary_data = { "id": summary_id, "company_id": company_id, "summary": summary_text, "created_by": user_id, "created_at": datetime.utcnow().isoformat(), "updated_at": datetime.utcnow().isoformat() } SHARED_SUMMARIES[summary_id] = summary_data logger.info(f"Summary {summary_id} created by {user_id} for company {company_id}") trigger_notification(summary_data) return summary_data def get_summaries(user_id: str, company_id: str) -> List[Dict[str, Any]]: """ Retrieve all conversation summaries for a given company. Only 'manager' or 'admin' roles are authorized. """ role = get_user_role(user_id, company_id) if role not in ("manager", "admin"): raise PermissionError("User not authorized to view summaries.") # Return only summaries that belong to the given company. return [s for s in SHARED_SUMMARIES.values() if s["company_id"] == company_id] def update_summary(user_id: str, company_id: str, summary_id: str, new_summary: str) -> Dict[str, Any]: """ Update an existing conversation summary. Only the creator or an 'admin' can update the summary. """ if summary_id not in SHARED_SUMMARIES: raise ValueError("Summary not found.") summary_data = SHARED_SUMMARIES[summary_id] if summary_data["company_id"] != company_id: raise PermissionError("This summary does not belong to your company.") role = get_user_role(user_id, company_id) if role != "admin" and summary_data["created_by"] != user_id: raise PermissionError("User not authorized to update this summary.") summary_data["summary"] = new_summary summary_data["updated_at"] = datetime.utcnow().isoformat() SHARED_SUMMARIES[summary_id] = summary_data logger.info(f"Summary {summary_id} updated by {user_id} for company {company_id}") return summary_data def delete_summary(user_id: str, company_id: str, summary_id: str) -> Dict[str, Any]: """ Delete a conversation summary. Only the creator or an 'admin' can delete the summary. """ if summary_id not in SHARED_SUMMARIES: raise ValueError("Summary not found.") summary_data = SHARED_SUMMARIES[summary_id] if summary_data["company_id"] != company_id: raise PermissionError("This summary does not belong to your company.") role = get_user_role(user_id, company_id) if role != "admin" and summary_data["created_by"] != user_id: raise PermissionError("User not authorized to delete this summary.") del SHARED_SUMMARIES[summary_id] logger.info(f"Summary {summary_id} deleted by {user_id} for company {company_id}") return {"status": "deleted", "id": summary_id} def trigger_notification(summary_data: Dict[str, Any]) -> None: """ Placeholder for real-time notifications. In production, integrate SSE or WebSocket to notify connected users. """ logger.info(f"Real-time notification triggered for summary {summary_data['id']}") # Integration with an SSE/WebSocket server would go here.