751 lines
34 KiB
Python
751 lines
34 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Cauldron Cloud — MCP Server
|
||
|
||
DEPRECATED (dismissed): this standalone stdio client was a test and is no longer
|
||
maintained. Development continues only on the HTTP MCP transport, served at
|
||
https://cauldron.cloud/mcp/<API_KEY> (src/Controllers/McpServerController.php in
|
||
the main cauldron repo). Kept here for reference only.
|
||
|
||
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)
|
||
|
||
|
||
# Filter params shared by list_deals and deal_statistics, so the same filter set
|
||
# is combinable with any listing or grouping.
|
||
_DEAL_STR_FILTERS = (
|
||
"firm", "stage", "status", "deal_type", "industry", "industry_category",
|
||
"country", "target_industry", "target_country",
|
||
"buyer_industry", "buyer_country", "seller_industry", "seller_country",
|
||
"pending_rfh",
|
||
"creation_from", "creation_to", "closing_from", "closing_to",
|
||
"updated_from", "updated_to",
|
||
)
|
||
_DEAL_INT_FILTERS = ("year_from", "year_to", "closing_year_from", "closing_year_to")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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 (partial match on Target/Seller/Buyer industry name). E.g. 'Automotive', 'Tech'.",
|
||
},
|
||
"industry_category": {
|
||
"type": "string",
|
||
"description": "Filter by Industry Category — the sector level above Industry (partial match on Target/Seller/Buyer). E.g. 'Industrials', 'Consumer'.",
|
||
},
|
||
"country": {
|
||
"type": "string",
|
||
"description": "Filter by country (partial match on Target/Seller/Buyer country). E.g. 'Italy', 'Germany'.",
|
||
},
|
||
"firm": {
|
||
"type": "string",
|
||
"description": "Filter by advisor firm name (partial match). E.g. 'Scouting', 'Quore'.",
|
||
},
|
||
"buyer_industry": {
|
||
"type": "string",
|
||
"description": "Filter specifically by the Buyer's industry (partial match).",
|
||
},
|
||
"buyer_country": {
|
||
"type": "string",
|
||
"description": "Filter specifically by the Buyer's country (partial match).",
|
||
},
|
||
"seller_industry": {
|
||
"type": "string",
|
||
"description": "Filter specifically by the Seller's industry (partial match).",
|
||
},
|
||
"seller_country": {
|
||
"type": "string",
|
||
"description": "Filter specifically by the Seller's country (partial match).",
|
||
},
|
||
"pending_rfh": {
|
||
"type": "string",
|
||
"enum": ["pending", "closed", "any"],
|
||
"description": "Filter by Request For Help state: 'pending' = open RFH, 'closed' = closed RFH, 'any' = has any RFH (open or closed).",
|
||
},
|
||
"search": {
|
||
"type": "string",
|
||
"description": "Free-text search across TargetName, BuyerName, SellerName, MandateDescription.",
|
||
},
|
||
"year_from": {
|
||
"type": "integer",
|
||
"description": "Filter deals whose CREATION year is >= this.",
|
||
},
|
||
"year_to": {
|
||
"type": "integer",
|
||
"description": "Filter deals whose CREATION year is <= this.",
|
||
},
|
||
"closing_year_from": {
|
||
"type": "integer",
|
||
"description": "Filter deals whose CLOSING year is >= this.",
|
||
},
|
||
"closing_year_to": {
|
||
"type": "integer",
|
||
"description": "Filter deals whose CLOSING year is <= this.",
|
||
},
|
||
"creation_from": {
|
||
"type": "string",
|
||
"description": "Creation date range start (YYYY-MM-DD).",
|
||
},
|
||
"creation_to": {
|
||
"type": "string",
|
||
"description": "Creation date range end (YYYY-MM-DD).",
|
||
},
|
||
"closing_from": {
|
||
"type": "string",
|
||
"description": "Closing date range start (YYYY-MM-DD).",
|
||
},
|
||
"closing_to": {
|
||
"type": "string",
|
||
"description": "Closing date range end (YYYY-MM-DD).",
|
||
},
|
||
"updated_from": {
|
||
"type": "string",
|
||
"description": "Last-update date range start (YYYY-MM-DD).",
|
||
},
|
||
"updated_to": {
|
||
"type": "string",
|
||
"description": "Last-update date range end (YYYY-MM-DD).",
|
||
},
|
||
"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 counts (and non-confidential total deal value) on visible "
|
||
"deals, grouped by one or two dimensions, with the same filters as list_deals so "
|
||
"any filter is combinable with any grouping.\n\n"
|
||
"Single dimension → rows of {label, count, total_value}.\n"
|
||
"Two dimensions (group_by_2) → cross-tab in long format: rows of "
|
||
"{label, label2, count, total_value} — ideal to then pivot (e.g. year × firm).\n\n"
|
||
"Useful for questions like:\n"
|
||
"- 'How many deals did we close per year?' (group_by='year_closed')\n"
|
||
"- 'Deals by firm and year' (group_by='year', group_by_2='firm')\n"
|
||
"- 'Crossborder vs domestic by industry category' (group_by='crossborder', group_by_2='industry_category')\n"
|
||
"- 'Buy-side deals by seller country in 2024' (group_by='seller_country', deal_type='Buy', year_from=2024, year_to=2024)"
|
||
),
|
||
inputSchema={
|
||
"type": "object",
|
||
"properties": {
|
||
"group_by": {
|
||
"type": "string",
|
||
"enum": [
|
||
"stage", "industry", "industry_category", "deal_type",
|
||
"year", "year_closed", "month", "firm", "country",
|
||
"target_industry", "target_country",
|
||
"buyer_industry", "buyer_country",
|
||
"seller_industry", "seller_country", "crossborder",
|
||
],
|
||
"description": (
|
||
"Primary dimension. 'year'/'month' = by creation date; 'year_closed' = by closing date; "
|
||
"'country' = target country; 'industry_category' = sector above industry. Default: 'stage'."
|
||
),
|
||
"default": "stage",
|
||
},
|
||
"group_by_2": {
|
||
"type": "string",
|
||
"enum": [
|
||
"stage", "industry", "industry_category", "deal_type",
|
||
"year", "year_closed", "month", "firm", "country",
|
||
"target_industry", "target_country",
|
||
"buyer_industry", "buyer_country",
|
||
"seller_industry", "seller_country", "crossborder",
|
||
],
|
||
"description": "Optional second dimension for a cross-tab (e.g. group_by='year', group_by_2='firm').",
|
||
},
|
||
"firm": {"type": "string", "description": "Filter by advisor firm name (partial match)."},
|
||
"stage": {"type": "string", "description": "Filter by deal stage (partial match)."},
|
||
"status": {
|
||
"type": "string",
|
||
"enum": ["open", "closed", "all"],
|
||
"description": "'open' = not Closed/Lost; 'closed' = Closed/Lost.",
|
||
},
|
||
"deal_type": {"type": "string", "description": "Filter by deal type (partial match). E.g. 'Sell', 'Buy'."},
|
||
"industry": {"type": "string", "description": "Filter by industry name (Target/Seller/Buyer, partial match)."},
|
||
"industry_category": {"type": "string", "description": "Filter by Industry Category / sector (Target/Seller/Buyer, partial match)."},
|
||
"country": {"type": "string", "description": "Filter by country (Target/Seller/Buyer, partial match)."},
|
||
"buyer_industry": {"type": "string", "description": "Filter by the Buyer's industry (partial match)."},
|
||
"buyer_country": {"type": "string", "description": "Filter by the Buyer's country (partial match)."},
|
||
"seller_industry": {"type": "string", "description": "Filter by the Seller's industry (partial match)."},
|
||
"seller_country": {"type": "string", "description": "Filter by the Seller's country (partial match)."},
|
||
"pending_rfh": {
|
||
"type": "string",
|
||
"enum": ["pending", "closed", "any"],
|
||
"description": "Filter by Request For Help state.",
|
||
},
|
||
"year_from": {"type": "integer", "description": "Creation year >= this."},
|
||
"year_to": {"type": "integer", "description": "Creation year <= this."},
|
||
"closing_year_from": {"type": "integer", "description": "Closing year >= this."},
|
||
"closing_year_to": {"type": "integer", "description": "Closing year <= this."},
|
||
"creation_from": {"type": "string", "description": "Creation date range start (YYYY-MM-DD)."},
|
||
"creation_to": {"type": "string", "description": "Creation date range end (YYYY-MM-DD)."},
|
||
"closing_from": {"type": "string", "description": "Closing date range start (YYYY-MM-DD)."},
|
||
"closing_to": {"type": "string", "description": "Closing date range end (YYYY-MM-DD)."},
|
||
"updated_from": {"type": "string", "description": "Last-update date range start (YYYY-MM-DD)."},
|
||
"updated_to": {"type": "string", "description": "Last-update date range end (YYYY-MM-DD)."},
|
||
},
|
||
},
|
||
),
|
||
|
||
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:
|
||
gb = meta["group_by"]
|
||
if meta.get("group_by_2"):
|
||
gb += " × " + meta["group_by_2"]
|
||
summary_parts.append(f"Grouped by: {gb}")
|
||
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_STR_FILTERS + ("search",):
|
||
if arguments.get(key):
|
||
params[key] = arguments[key]
|
||
for key in _DEAL_INT_FILTERS + ("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: dict[str, Any] = {}
|
||
for key in ("group_by", "group_by_2"):
|
||
if arguments.get(key):
|
||
params[key] = arguments[key]
|
||
for key in _DEAL_STR_FILTERS:
|
||
if arguments.get(key):
|
||
params[key] = arguments[key]
|
||
for key in _DEAL_INT_FILTERS:
|
||
if arguments.get(key) is not None:
|
||
params[key] = int(arguments[key])
|
||
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())
|