#!/usr/bin/env python3 """ Cauldron Cloud — MCP Server Connects Claude Desktop (or any MCP client) to the Cauldron portal REST API. Setup: pip install mcp httpx export CAULDRON_API_KEY=cldrn_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx export CAULDRON_API_URL=https://cauldron.cloud/api # optional override Claude Desktop config (~/.config/Claude/claude_desktop_config.json on Linux, ~/Library/Application Support/Claude/claude_desktop_config.json on Mac): { "mcpServers": { "cauldron": { "command": "python", "args": ["/path/to/cauldron-mcp/server.py"], "env": { "CAULDRON_API_KEY": "cldrn_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx" } } } } """ import os import sys import json import httpx import asyncio from typing import Any from mcp.server import Server from mcp.server.stdio import stdio_server from mcp import types # --------------------------------------------------------------------------- # Config # --------------------------------------------------------------------------- API_KEY = os.environ.get("CAULDRON_API_KEY", "") API_URL = os.environ.get("CAULDRON_API_URL", "https://cauldron.cloud/api").rstrip("/") if not API_KEY: print( "ERROR: CAULDRON_API_KEY environment variable is required.\n" "Set it in your Claude Desktop MCP configuration.", file=sys.stderr, ) sys.exit(1) # --------------------------------------------------------------------------- # HTTP client helper # --------------------------------------------------------------------------- def _headers() -> dict: return {"X-API-Key": API_KEY, "Accept": "application/json"} async def _get(path: str, params: dict | None = None) -> dict: """Make a GET request to the Cauldron API and return the parsed JSON.""" url = f"{API_URL}{path}" async with httpx.AsyncClient(timeout=30) as client: resp = await client.get(url, headers=_headers(), params=params or {}) try: data = resp.json() except Exception: return {"success": False, "error": f"HTTP {resp.status_code}: non-JSON response"} if resp.status_code == 401: return {"success": False, "error": "Invalid or revoked API key. Check CAULDRON_API_KEY."} if resp.status_code == 403: return {"success": False, "error": "Access denied: " + data.get("error", "")} if resp.status_code == 404: return {"success": False, "error": "Not found: " + data.get("error", "")} if not resp.is_success: return {"success": False, "error": data.get("error", f"HTTP {resp.status_code}")} return data def _fmt(data: Any) -> str: """Format API response as a readable string for the LLM.""" if isinstance(data, (dict, list)): return json.dumps(data, indent=2, ensure_ascii=False, default=str) return str(data) # --------------------------------------------------------------------------- # MCP Server # --------------------------------------------------------------------------- app = Server("cauldron-cloud") @app.list_tools() async def list_tools() -> list[types.Tool]: return [ types.Tool( name="get_my_profile", description=( "Returns information about the authenticated user: name, email, " "role (profile), and the firm they belong to. " "Use this to understand the current user's context before answering questions." ), inputSchema={"type": "object", "properties": {}, "required": []}, ), types.Tool( name="list_deals", description=( "Lists M&A and financing deals visible to the user. " "Only returns active deals (deleted ones are excluded). " "A user sees deals from their own firm plus all shared deals from the network.\n\n" "Use 'status=open' for ongoing deals (Pitch/Mandate/Signed/On-hold). " "Use 'status=closed' for completed or lost deals.\n\n" "deal_type examples: 'Sell Side', 'Buy Side', 'Financing', 'ECM', 'DCM', 'Fairness Opinion'.\n" "For deals where a company is SELLING and looking for a buyer → deal_type='Sell Side'.\n" "For deals where a client wants to ACQUIRE a company → deal_type='Buy Side'.\n" "industry examples: 'Automotive', 'Technology', 'Healthcare', 'Consumer'.\n" "stage examples: 'Pitch', 'Mandate', 'Signed', 'Closed', 'Lost', 'On-hold'." ), inputSchema={ "type": "object", "properties": { "deal_type": { "type": "string", "description": "Filter by deal type (partial match). E.g. 'Sell Side', 'Buy Side', 'Financing'.", }, "stage": { "type": "string", "description": "Filter by deal stage (partial match). E.g. 'Mandate', 'Closed', 'Pitch'.", }, "status": { "type": "string", "enum": ["open", "closed", "all"], "description": "'open' = Pitch/Mandate/Signed/On-hold; 'closed' = Closed/Lost; 'all' = no filter.", "default": "all", }, "industry": { "type": "string", "description": "Filter by industry sector (partial match on Target/Seller/Buyer). E.g. 'Automotive', 'Tech'.", }, "country": { "type": "string", "description": "Filter by country (partial match on Target/Seller/Buyer country). E.g. 'Italy', 'Germany'.", }, "search": { "type": "string", "description": "Free-text search across TargetName, BuyerName, SellerName, MandateDescription.", }, "year_from": { "type": "integer", "description": "Filter deals created from this year onwards.", }, "year_to": { "type": "integer", "description": "Filter deals created up to this year.", }, "shared_only": { "type": "boolean", "description": "If true, return only deals shared across the network.", }, "limit": { "type": "integer", "description": "Max number of results (1–100, default 20).", "default": 20, }, "offset": { "type": "integer", "description": "Pagination offset (default 0).", "default": 0, }, }, }, ), types.Tool( name="get_deal_detail", description=( "Returns the full details of a single deal by its ID, including " "all parties (Target, Seller, Buyer), financial data (confidential values are masked), " "stage, countries, industries, ownership types, contacts, and staff involved. " "Use list_deals first to find the deal ID." ), inputSchema={ "type": "object", "properties": { "deal_id": { "type": "integer", "description": "The numeric ID of the deal (IdDeal).", } }, "required": ["deal_id"], }, ), types.Tool( name="deal_statistics", description=( "Returns aggregated statistics on visible deals, grouped by a chosen dimension. " "Useful for analytics questions like:\n" "- 'How many deals did we close per year?'\n" "- 'Which industry has the most deals?'\n" "- 'What is our deal breakdown by stage?'" ), inputSchema={ "type": "object", "properties": { "group_by": { "type": "string", "enum": ["stage", "industry", "deal_type", "year", "month", "firm", "country"], "description": "Dimension to group by. 'month' = per month (YYYY-MM). 'country' = target country. Default: 'stage'.", "default": "stage", }, "year_from": { "type": "integer", "description": "Include deals from this year onwards.", }, "year_to": { "type": "integer", "description": "Include deals up to this year.", }, }, }, ), types.Tool( name="list_my_requests_for_help", description=( "Lists Request for Help (RFH) that MY FIRM has sent to the network. " "These are requests we created asking other firms for buy-side leads or suggestions. " "Shows how many responses each request has received.\n\n" "Use status='open' for pending requests, 'closed' for completed ones." ), inputSchema={ "type": "object", "properties": { "status": { "type": "string", "enum": ["open", "closed", "all"], "description": "Filter by request status. Default: 'all'.", "default": "all", }, "deal_id": { "type": "integer", "description": "Optional: filter by a specific deal ID.", }, }, }, ), types.Tool( name="list_incoming_requests", description=( "Lists Request for Help (RFH) sent by OTHER firms that are directed to MY FIRM, " "asking us for buy-side leads or suggestions. " "By default returns only requests we haven't answered yet.\n\n" "Example use: 'Show me the requests for help we haven't responded to yet'\n" "Example use: 'Are there any open RFH we should answer?'" ), inputSchema={ "type": "object", "properties": { "unanswered_only": { "type": "boolean", "description": "If true (default), returns only requests with no response from our firm yet.", "default": True, }, "status": { "type": "string", "enum": ["open", "closed", "all"], "description": "Filter by request status. Default: 'open'.", "default": "open", }, }, }, ), types.Tool( name="list_deal_stages", description="Returns the list of available deal stages (Pitch, Mandate, Signed, Closed, On-hold, Lost).", inputSchema={"type": "object", "properties": {}, "required": []}, ), types.Tool( name="list_industries", description="Returns the list of industry classifications available on the platform. Optionally filter by sector.", inputSchema={ "type": "object", "properties": { "sector": { "type": "string", "description": "Optional partial match on the sector name.", } }, }, ), types.Tool( name="list_firms", description=( "Returns the firms visible to the current user: " "their own firm plus all firms that have at least one deal shared with the network. " "Each firm includes a summary description, countries of activity, and equity partner info." ), inputSchema={"type": "object", "properties": {}, "required": []}, ), types.Tool( name="get_deals_rich", description=( "Returns a rich dataset of deals using the portal's export engine. " "Includes fields NOT available in list_deals:\n" "- TransactionTitle and TransactionDescription: the tombstone narrative text\n" "- Quote and QuoteAuthor: a client or deal quote\n" "- DealAttachments: list of attached document filenames\n" "- SuccessFee, FeeShare: fee structure\n" "- Multiple, MultipleOverride: valuation multiples\n" "- CoAdvisory, OtherFirmInDeal, OtherFirmInvolved: network co-advisory relationships\n" "- Financing-specific fields (PurposeOfFinancing, TypeOfLending, etc.)\n\n" "Confidential financial values are masked. HTML is stripped from narrative fields.\n\n" "Use this when you need the full deal narrative, tombstone text, or fee/multiple data. " "Use list_deals for quick filtered searches." ), inputSchema={ "type": "object", "properties": { "date_from": { "type": "string", "description": "Start of date range (YYYY-MM-DD). Defaults to 5 years ago.", }, "date_to": { "type": "string", "description": "End of date range (YYYY-MM-DD). Defaults to today.", }, }, }, ), types.Tool( name="list_members", description=( "Lists people (professionals) visible to the current user: members of their own firm " "plus members of other firms that have at least one shared deal in the network.\n\n" "Useful for questions like:\n" "- 'Who covers Automotive deals in Germany?'\n" "- 'List all partners in the network'\n" "- 'Who are the directors at [FirmName]?'\n" "- 'Find me someone who covers Healthcare in Italy'\n\n" "Each member includes their firm, job title, city, sectors they cover, and countries." ), inputSchema={ "type": "object", "properties": { "search": { "type": "string", "description": "Free-text search on name or email.", }, "firm_id": { "type": "integer", "description": "Filter by firm ID (use list_firms to get IDs).", }, "job_title": { "type": "string", "description": "Partial match on job title. E.g. 'Partner', 'Director', 'Associate'.", }, "staff_category": { "type": "string", "description": "Partial match on staff category. E.g. 'partners', 'directors', 'associates'.", }, "sector": { "type": "string", "description": "Filter by industry sector covered. E.g. 'Automotive', 'Healthcare', 'Technology'.", }, "country": { "type": "string", "description": "Filter by country of activity. E.g. 'Italy', 'Germany', 'France'.", }, "limit": { "type": "integer", "description": "Max results (1–100, default 30).", "default": 30, }, "offset": { "type": "integer", "description": "Pagination offset (default 0).", "default": 0, }, }, }, ), types.Tool( name="get_member_detail", description=( "Returns the full profile of a single network member by their ID, including " "job title, city, phone numbers, firm, all sectors covered, countries of activity, " "and their professional summary." ), inputSchema={ "type": "object", "properties": { "member_id": { "type": "integer", "description": "The numeric ID of the member (from list_members).", } }, "required": ["member_id"], }, ), types.Tool( name="list_rfh_suggestions", description=( "Returns all suggestions (buy-side leads / candidates) submitted for a specific " "Request for Help (RFH).\n\n" "If the RFH was created by YOUR firm: you see all suggestions from all responding firms.\n" "If it's an incoming RFH: you see only your firm's own submissions.\n\n" "Each suggestion includes: suggesting firm, status (New/Go/No Go/Read), " "whether it was selected, company info, financials (Turnover, EBITDA), and reasoning.\n\n" "Use list_my_requests_for_help or list_incoming_requests to find RFH IDs first." ), inputSchema={ "type": "object", "properties": { "rfh_id": { "type": "integer", "description": "The ID of the Request for Help (IdRequestForHelp).", } }, "required": ["rfh_id"], }, ), types.Tool( name="rfh_statistics", description=( "Returns aggregated statistics on Requests for Help (RFH) activity for the current firm.\n\n" "Covers two dimensions:\n" "1. **Outgoing RFH** (requests my firm sent): how many, open vs closed, " "how many received no response, average suggestions received, " "breakdown by suggestion status (Go/No Go/New/Read), selected suggestions, " "and which firms in the network respond most actively.\n\n" "2. **Incoming RFH** (requests from other firms targeting my firm): " "how many we've been asked, how many we haven't answered yet, " "our response rate, average days to respond, " "and breakdown of our suggestions by status.\n\n" "Example questions:\n" "- 'How are our Requests for Help performing?'\n" "- 'Which firms send us the most suggestions?'\n" "- 'How many incoming RFH have we not answered yet?'\n" "- 'What is our average response time on incoming requests?'" ), inputSchema={"type": "object", "properties": {}, "required": []}, ), ] @app.call_tool() async def call_tool(name: str, arguments: dict) -> list[types.TextContent]: """Dispatch tool calls to the Cauldron API.""" async def respond(result: dict) -> list[types.TextContent]: if not result.get("success", True): text = f"Error: {result.get('error', 'Unknown error')}" else: data = result.get("data", result) meta = result.get("meta", {}) parts = [] if meta: summary_parts = [] if "total" in meta: summary_parts.append(f"Total: {meta['total']}, returned: {meta.get('returned', '?')}") if "firm" in meta: summary_parts.append(f"Firm: {meta['firm']}") if "group_by" in meta: summary_parts.append(f"Grouped by: {meta['group_by']}") if "unanswered_only" in meta: summary_parts.append(f"Unanswered only: {meta['unanswered_only']}") if summary_parts: parts.append("# " + " | ".join(summary_parts)) parts.append(_fmt(data)) text = "\n".join(parts) return [types.TextContent(type="text", text=text)] # ------------------------------------------------------------------ if name == "get_my_profile": result = await _get("/me") return await respond(result) # ------------------------------------------------------------------ elif name == "list_deals": params: dict[str, Any] = {} for key in ("deal_type", "stage", "status", "industry", "country", "search"): if arguments.get(key): params[key] = arguments[key] for key in ("year_from", "year_to", "limit", "offset"): if arguments.get(key) is not None: params[key] = int(arguments[key]) if arguments.get("shared_only"): params["shared_only"] = "1" result = await _get("/deals", params) return await respond(result) # ------------------------------------------------------------------ elif name == "get_deal_detail": deal_id = int(arguments["deal_id"]) result = await _get(f"/deal/{deal_id}") return await respond(result) # ------------------------------------------------------------------ elif name == "deal_statistics": params = {} if arguments.get("group_by"): params["group_by"] = arguments["group_by"] if arguments.get("year_from"): params["year_from"] = int(arguments["year_from"]) if arguments.get("year_to"): params["year_to"] = int(arguments["year_to"]) result = await _get("/stats", params) return await respond(result) # ------------------------------------------------------------------ elif name == "list_my_requests_for_help": params = {} if arguments.get("status"): params["status"] = arguments["status"] if arguments.get("deal_id"): params["deal_id"] = int(arguments["deal_id"]) result = await _get("/my-requests", params) return await respond(result) # ------------------------------------------------------------------ elif name == "list_incoming_requests": params: dict[str, Any] = {} unanswered = arguments.get("unanswered_only", True) params["unanswered_only"] = "0" if unanswered is False else "1" if arguments.get("status"): params["status"] = arguments["status"] result = await _get("/incoming-requests", params) return await respond(result) # ------------------------------------------------------------------ elif name == "list_deal_stages": result = await _get("/stages") return await respond(result) # ------------------------------------------------------------------ elif name == "list_industries": params = {} if arguments.get("sector"): params["sector"] = arguments["sector"] result = await _get("/industries", params) return await respond(result) # ------------------------------------------------------------------ elif name == "list_firms": result = await _get("/firms") return await respond(result) # ------------------------------------------------------------------ elif name == "get_deals_rich": params: dict[str, Any] = {} if arguments.get("date_from"): params["date_from"] = arguments["date_from"] if arguments.get("date_to"): params["date_to"] = arguments["date_to"] result = await _get("/deal-export", params) return await respond(result) # ------------------------------------------------------------------ elif name == "list_members": params: dict[str, Any] = {} for key in ("search", "job_title", "staff_category", "sector", "country"): if arguments.get(key): params[key] = arguments[key] for key in ("firm_id", "limit", "offset"): if arguments.get(key) is not None: params[key] = int(arguments[key]) result = await _get("/members", params) return await respond(result) # ------------------------------------------------------------------ elif name == "get_member_detail": member_id = int(arguments["member_id"]) result = await _get(f"/member/{member_id}") return await respond(result) # ------------------------------------------------------------------ elif name == "list_rfh_suggestions": rfh_id = int(arguments["rfh_id"]) result = await _get(f"/rfh/{rfh_id}/suggestions") return await respond(result) # ------------------------------------------------------------------ elif name == "rfh_statistics": result = await _get("/rfh-stats") return await respond(result) # ------------------------------------------------------------------ else: return [types.TextContent(type="text", text=f"Unknown tool: {name}")] # --------------------------------------------------------------------------- # Entry point # --------------------------------------------------------------------------- async def main(): async with stdio_server() as (read_stream, write_stream): await app.run(read_stream, write_stream, app.create_initialization_options()) if __name__ == "__main__": asyncio.run(main())