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