commit 20e241f925b79baab6b8fc6ca8d04408e1c6994e Author: Spaike Date: Thu May 7 21:54:24 2026 +0200 Initial release: Cauldron Cloud MCP server Connects Claude Desktop (or any MCP client) to the Cauldron portal REST API. Supports deal listing, filtering, statistics, Request for Help tracking, and firm/industry lookup — all firm-scoped with the same permission model as the web portal. diff --git a/README.md b/README.md new file mode 100644 index 0000000..baa19ec --- /dev/null +++ b/README.md @@ -0,0 +1,110 @@ +# Cauldron Cloud — MCP Server + +Collega Claude Desktop (o qualsiasi client MCP) al portale Cauldron. + +## Prerequisiti + +- Python 3.10+ +- Pip + +## Installazione + +```bash +pip install mcp httpx +``` + +Oppure con un virtual environment (consigliato): + +```bash +python -m venv .venv +source .venv/bin/activate # Linux/Mac +# oppure: .venv\Scripts\activate # Windows +pip install -r requirements.txt +``` + +## Ottenere la propria API Key + +Accedere al portale Cauldron → Profilo → API Keys → "Genera nuova chiave". +La chiave ha il formato: `cldrn_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx` + +> La chiave viene mostrata UNA SOLA VOLTA. Conservarla in modo sicuro. + +## Configurazione Claude Desktop + +Aprire il file di configurazione di Claude Desktop: + +- **Mac**: `~/Library/Application Support/Claude/claude_desktop_config.json` +- **Linux**: `~/.config/Claude/claude_desktop_config.json` +- **Windows**: `%APPDATA%\Claude\claude_desktop_config.json` + +Aggiungere (o completare) la sezione `mcpServers`: + +```json +{ + "mcpServers": { + "cauldron": { + "command": "python", + "args": ["/percorso/assoluto/al/server.py"], + "env": { + "CAULDRON_API_KEY": "cldrn_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx" + } + } + } +} +``` + +Sostituire `/percorso/assoluto/al/server.py` con il path reale del file. +Se si usa un venv: `"command": "/percorso/.venv/bin/python"`. + +Riavviare Claude Desktop per caricare la configurazione. + +## Tool disponibili + +| Tool | Descrizione | +|---|---| +| `get_my_profile` | Info sull'utente autenticato e la sua firma | +| `list_deals` | Lista deal con filtri (tipo, stage, industria, paese, testo libero) | +| `get_deal_detail` | Dettaglio completo di un deal | +| `deal_statistics` | Statistiche aggregate (per stage, industria, anno, ecc.) | +| `list_my_requests_for_help` | RFH inviati dalla mia firma al network | +| `list_incoming_requests` | RFH di altre firm diretti a noi (con filtro "senza risposta") | +| `list_deal_stages` | Elenco stage disponibili | +| `list_industries` | Elenco classificazioni industry | +| `list_firms` | Elenco firm visibili | + +## Esempi di domande a Claude + +``` +"Mostrami tutti i deal Sell Side ancora aperti nel settore Automotive" + +"Quanti deal abbiamo chiuso nel 2024 per industria?" + +"Ci sono Request for Help a cui non abbiamo ancora risposto?" + +"Dammi il dettaglio del deal #142" + +"Quali deal condivisi nel network riguardano il settore Technology?" + +"Statistiche sui nostri deal per anno, dal 2022 ad oggi" +``` + +## Sicurezza + +- La API key identifica univocamente l'utente — trattarla come una password +- La visibilità è identica al portale web: solo i propri deal + quelli condivisi +- I deal cancellati non vengono mai restituiti +- I valori finanziari con flag di confidenzialità vengono mascherati automaticamente +- La chiave è revocabile in qualsiasi momento dal portale + +## Permessi MCP vs Portale + +| Aspetto | Portale web | MCP | +|---|---|---| +| Admin Portale vede tutti i deal | ✅ | ❌ (firma-scoped) | +| Deal propria firma | ✅ | ✅ | +| Deal condivisi altre firm | ✅ | ✅ | +| Deal cancellati (Attivo=N) | ❌ | ❌ | +| Valori confidenziali | Visibili se autorizzato | Mascherati | + +> Il layer MCP è intenzionalmente più restrittivo: nessun utente ottiene visibilità +> super-admin attraverso l'API key, indipendentemente dal proprio ruolo sul portale. diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..9714bd9 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,2 @@ +mcp>=1.0.0 +httpx>=0.27.0 diff --git a/server.py b/server.py new file mode 100644 index 0000000..e14a742 --- /dev/null +++ b/server.py @@ -0,0 +1,428 @@ +#!/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", "firm"], + "description": "Dimension to group by. 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." + ), + 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) + + # ------------------------------------------------------------------ + 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())