Spaike
/
cauldron-mcp
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Cauldron Cloud MCP Server — initial release 

Connects Claude Desktop to the Cauldron M&A portal via MCP.
Requires Python 3.10+ and an API key from the portal.
This commit is contained in:
Spaike 2026-05-08 17:24:29 +02:00
commit ebe1c78ef1
4 changed files with 895 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

109
README.md Normal file
View File

@ -0,0 +1,109 @@
# Cauldron Cloud — MCP Server
Connects Claude Desktop (or any MCP client) to the Cauldron M&A portal.
## Quick setup — 3 commands
```bash
git clone https://git.zerotohero.it/Spaike/cauldron-mcp.git
cd cauldron-mcp
python setup.py
```
The wizard will:
1. Install dependencies automatically
2. Open the browser to the API Keys page on the portal
3. Read the key you paste into the terminal
4. Write `claude_desktop_config.json` without touching your other settings
Then restart Claude Desktop — done.
---
## Manual setup
### Requirements
- Python 3.10+
- pip
### Install dependencies
```bash
pip install mcp httpx
```
### Get your API key
Go to the Cauldron portal → user menu → **API Keys** → click **Generate**.
Your key looks like: `cldrn_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx`
> The key is shown **once only**. Store it somewhere safe.
### Configure Claude Desktop
Open the config file:
- **Mac**: `~/Library/Application Support/Claude/claude_desktop_config.json`
- **Linux**: `~/.config/Claude/claude_desktop_config.json`
- **Windows**: `%APPDATA%\Claude\claude_desktop_config.json`
Add the `mcpServers` block:
```json
{
"mcpServers": {
"cauldron": {
"command": "python",
"args": ["/absolute/path/to/cauldron-mcp/server.py"],
"env": {
"CAULDRON_API_KEY": "cldrn_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
}
}
}
}
```
Restart Claude Desktop.
---
## Available tools
| Tool | Description |
|---|---|
| `get_my_profile` | Your user profile and firm |
| `list_deals` | List deals with filters (type, stage, industry, country, free text) |
| `get_deal_detail` | Full detail of a single deal |
| `deal_statistics` | Aggregated stats grouped by stage, industry, year, month, country, etc. |
| `list_my_requests_for_help` | Requests for Help sent by your firm to the network |
| `list_incoming_requests` | Incoming RFH from other firms (filterable by unanswered) |
| `list_rfh_suggestions` | Suggestions submitted for a specific Request for Help |
| `rfh_statistics` | Aggregated RFH activity stats (response rates, top responders, etc.) |
| `list_members` | Network professionals (filter by sector, country, job title, firm) |
| `get_member_detail` | Full profile of a single network member |
| `list_deal_stages` | Available deal stages |
| `list_industries` | Industry classifications |
| `list_firms` | Firms visible to your account |
---
## Example questions for Claude
```
"Show me all open Sell Side deals in the Automotive sector"
"How many deals did we close in 2024, broken down by industry?"
"Are there any Requests for Help we haven't answered yet?"
"Give me the full detail of deal #142"
"What shared network deals are in the Technology sector?"
"Show deal statistics by month from 2023 to today"
"Who in the network covers Healthcare deals in Germany?"
"How is our firm performing on incoming Requests for Help?"
```

2
requirements.txt Normal file
View File

@ -0,0 +1,2 @@
mcp>=1.0.0
httpx>=0.27.0

574
server.py Normal file
View File

@ -0,0 +1,574 @@
#!/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."
),
inputSchema={"type": "object", "properties": {}, "required": []},
),
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 == "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())

210
setup.py Normal file
View File

@ -0,0 +1,210 @@
#!/usr/bin/env python3
"""
Cauldron Cloud MCP — Setup wizard
Installs dependencies and writes Claude Desktop config automatically.
"""
import json
import os
import platform
import shutil
import subprocess
import sys
import webbrowser
from pathlib import Path
# ── Colours ──────────────────────────────────────────────────────────────────
def _supports_colour() -> bool:
return sys.stdout.isatty() and platform.system() != "Windows"
if _supports_colour():
RESET = "\033[0m"
BOLD = "\033[1m"
GREEN = "\033[32m"
YELLOW = "\033[33m"
RED = "\033[31m"
CYAN = "\033[36m"
else:
RESET = BOLD = GREEN = YELLOW = RED = CYAN = ""
def ok(msg: str) -> None: print(f"{GREEN} ✓ {RESET}{msg}")
def info(msg: str) -> None: print(f"{CYAN} → {RESET}{msg}")
def warn(msg: str) -> None: print(f"{YELLOW} ⚠ {RESET}{msg}")
def err(msg: str) -> None: print(f"{RED} ✗ {RESET}{msg}")
def bold(msg: str) -> str: return f"{BOLD}{msg}{RESET}"
# ── Helpers ───────────────────────────────────────────────────────────────────
def _python_executable() -> str:
"""Return the Python executable that is running this script."""
return sys.executable
def _config_path() -> Path:
system = platform.system()
if system == "Windows":
base = Path(os.environ.get("APPDATA", "~")).expanduser()
return base / "Claude" / "claude_desktop_config.json"
if system == "Darwin":
return Path.home() / "Library" / "Application Support" / "Claude" / "claude_desktop_config.json"
# Linux / other
xdg = os.environ.get("XDG_CONFIG_HOME", "")
base = Path(xdg) if xdg else Path.home() / ".config"
return base / "Claude" / "claude_desktop_config.json"
def _read_config(path: Path) -> dict:
if path.exists():
try:
return json.loads(path.read_text(encoding="utf-8"))
except json.JSONDecodeError:
warn("Existing config file contains invalid JSON — it will be backed up and rewritten.")
backup = path.with_suffix(".json.bak")
shutil.copy(path, backup)
info(f"Backup saved to {backup}")
return {}
def _write_config(path: Path, data: dict) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
def _install_deps() -> bool:
packages = ["mcp>=1.0.0", "httpx>=0.27.0"]
info("Installing Python dependencies...")
result = subprocess.run(
[_python_executable(), "-m", "pip", "install", "--quiet", *packages],
capture_output=True,
text=True,
)
if result.returncode != 0:
err("pip install failed:")
print(result.stderr.strip())
return False
return True
def _check_deps() -> bool:
try:
import mcp # noqa: F401
import httpx # noqa: F401
return True
except ImportError:
return False
def _validate_key(key: str) -> bool:
key = key.strip()
return key.startswith("cldrn_") and len(key) >= 20
# ── Main ──────────────────────────────────────────────────────────────────────
def main() -> None:
print()
print(bold("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"))
print(bold(" Cauldron Cloud — MCP Server Setup"))
print(bold("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"))
print()
# ── 1. Python version check ───────────────────────────────────────────────
if sys.version_info < (3, 10):
err(f"Python 3.10+ required (you have {platform.python_version()}).")
sys.exit(1)
ok(f"Python {platform.python_version()}")
# ── 2. Dependencies ───────────────────────────────────────────────────────
if _check_deps():
ok("Dependencies already installed (mcp, httpx)")
else:
if not _install_deps():
err("Could not install dependencies. Try manually: pip install mcp httpx")
sys.exit(1)
if not _check_deps():
err("Dependencies installed but import still fails. Check your Python environment.")
sys.exit(1)
ok("Dependencies installed (mcp, httpx)")
# ── 3. Locate server.py ───────────────────────────────────────────────────
server_py = Path(__file__).resolve().parent / "server.py"
if not server_py.exists():
err(f"server.py not found at {server_py}")
sys.exit(1)
ok(f"Server script: {server_py}")
# ── 4. Get the API key ────────────────────────────────────────────────────
print()
print(bold(" Step: Generate your API key"))
print()
print(" A browser window will open to the Cauldron portal.")
print(" Log in, click Generate , copy the key and paste it below.")
print()
input(" Press ENTER to open the browser...")
portal_url = "https://cauldron.cloud/mcpKeys"
webbrowser.open(portal_url)
info(f"Opened: {portal_url}")
print()
api_key = ""
for attempt in range(3):
raw = input(" Paste your API key here: ").strip()
if _validate_key(raw):
api_key = raw
break
err("Key must start with cldrn_ and be at least 20 characters. Try again.")
else:
err("No valid key provided. Run setup.py again when you have a key.")
sys.exit(1)
ok(f"Key accepted: {api_key[:14]}…")
# ── 5. Detect Claude Desktop config ───────────────────────────────────────
config_path = _config_path()
print()
info(f"Config file: {config_path}")
config = _read_config(config_path)
# ── 6. Merge mcpServers entry ─────────────────────────────────────────────
python_cmd = _python_executable()
mcp_entry = {
"command": python_cmd,
"args": [str(server_py)],
"env": {"CAULDRON_API_KEY": api_key},
}
if "mcpServers" not in config:
config["mcpServers"] = {}
existing = config["mcpServers"].get("cauldron")
config["mcpServers"]["cauldron"] = mcp_entry
_write_config(config_path, config)
if existing:
ok("Updated existing cauldron entry in Claude Desktop config")
else:
ok("Added cauldron entry to Claude Desktop config")
# ── 7. Done ───────────────────────────────────────────────────────────────
print()
print(bold("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"))
print(bold(" Setup complete!"))
print(bold("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"))
print()
print(" Next steps:")
print(f" 1. {bold('Restart Claude Desktop')}")
print(" 2. Open a new Chat")
print(" 3. Look for the 🔌 icon — cauldron should appear in the tool list")
print()
print(" Try asking Claude:")
print(f" {CYAN}\"Show me all open Sell Side deals in the Automotive sector\"{RESET}")
print(f" {CYAN}\"Are there any incoming Requests for Help we haven't answered?\"{RESET}")
print()
if __name__ == "__main__":
main()