429 lines
17 KiB
Python
429 lines
17 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", "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())
|