server_all_tools.py•7.16 kB
"""
CATS MCP Server - All Tools Loaded at Import Time
"""
from __future__ import annotations
import os
import asyncio
from typing import Any, Optional, Callable, Awaitable
import httpx
from dotenv import load_dotenv
from fastmcp import FastMCP
load_dotenv()
# Configuration
CATS_API_BASE_URL = os.getenv("CATS_API_BASE_URL", "https://api.catsone.com/v3")
CATS_API_KEY = os.getenv("CATS_API_KEY", "")
mcp = FastMCP("CATS API v3")
class CATSAPIError(Exception):
"""CATS API error"""
pass
async def make_request(
method: str,
endpoint: str,
params: Optional[dict[str, Any]] = None,
json_data: Optional[dict[str, Any]] = None
) -> dict[str, Any]:
"""
Make authenticated request to CATS API with exponential backoff retry
Retry strategy:
- 429 (Rate Limit): Retry with exponential backoff (1s → 2s → 4s → 8s)
- 5xx (Server Error): Retry up to 3 times
- Other errors: Fail immediately
"""
if not CATS_API_KEY:
raise CATSAPIError("CATS_API_KEY not configured")
url = f"{CATS_API_BASE_URL.rstrip('/')}/{endpoint.lstrip('/')}"
headers = {
"Authorization": f"Token {CATS_API_KEY}",
"Content-Type": "application/json",
}
max_retries = 4
base_delay = 1.0 # seconds
for attempt in range(max_retries):
try:
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.request(method, url, headers=headers, params=params, json=json_data)
# Check for rate limiting
if response.status_code == 429:
if attempt < max_retries - 1:
delay = base_delay * (2 ** attempt) # Exponential backoff: 1s, 2s, 4s, 8s
print(f"⚠️ Rate limit hit, retrying in {delay}s (attempt {attempt + 1}/{max_retries})")
await asyncio.sleep(delay)
continue
else:
raise CATSAPIError("Rate limit exceeded after max retries")
# Check for server errors (5xx)
if 500 <= response.status_code < 600:
if attempt < max_retries - 1:
delay = base_delay * (2 ** attempt)
print(f"⚠️ Server error {response.status_code}, retrying in {delay}s")
await asyncio.sleep(delay)
continue
# Raise for other HTTP errors
response.raise_for_status()
return response.json()
except httpx.TimeoutException as e:
if attempt < max_retries - 1:
delay = base_delay * (2 ** attempt)
print(f"⚠️ Timeout, retrying in {delay}s")
await asyncio.sleep(delay)
continue
raise CATSAPIError(f"Request timeout after {max_retries} attempts: {e}")
except httpx.HTTPStatusError as e:
# Don't retry on client errors (4xx except 429)
if 400 <= e.response.status_code < 500 and e.response.status_code != 429:
raise CATSAPIError(f"API error: {e}")
# Retry on network errors
if attempt < max_retries - 1:
delay = base_delay * (2 ** attempt)
print(f"⚠️ Network error, retrying in {delay}s")
await asyncio.sleep(delay)
continue
raise CATSAPIError(f"API error after {max_retries} attempts: {e}")
except httpx.HTTPError as e:
# Retry on other HTTP errors (network issues, etc.)
if attempt < max_retries - 1:
delay = base_delay * (2 ** attempt)
print(f"⚠️ HTTP error, retrying in {delay}s")
await asyncio.sleep(delay)
continue
raise CATSAPIError(f"HTTP error after {max_retries} attempts: {e}")
# Should never reach here, but satisfy type checker
raise CATSAPIError(f"Request failed after {max_retries} attempts")
# LOAD ALL TOOLSETS AT MODULE IMPORT TIME
print("Loading all CATS toolsets...")
from toolsets_default import (
register_candidates_tools,
register_jobs_tools,
register_pipelines_tools,
register_context_tools,
register_tasks_tools
)
from toolsets_recruiting import (
register_companies_tools,
register_contacts_tools,
register_activities_tools,
register_portals_tools,
register_work_history_tools
)
from toolsets_data import (
register_tags_tools,
register_webhooks_tools,
register_users_tools,
register_triggers_tools,
register_attachments_tools,
register_backups_tools,
register_events_tools
)
# Import prompts, resources, and templates
from prompts_recruiting import (
register_recruiting_prompts,
register_interview_prompts
)
from resources_candidates import (
register_candidate_resources,
register_job_resources
)
from templates_emails import (
register_email_templates
)
# Register all toolsets immediately
register_candidates_tools(mcp, make_request)
print(" ✓ candidates (28 tools)")
register_jobs_tools(mcp, make_request)
print(" ✓ jobs (40 tools)")
register_pipelines_tools(mcp, make_request)
print(" ✓ pipelines (13 tools)")
register_context_tools(mcp, make_request)
print(" ✓ context (3 tools)")
register_tasks_tools(mcp, make_request)
print(" ✓ tasks (5 tools)")
register_companies_tools(mcp, make_request)
print(" ✓ companies (18 tools)")
register_contacts_tools(mcp, make_request)
print(" ✓ contacts (18 tools)")
register_activities_tools(mcp, make_request)
print(" ✓ activities (6 tools)")
register_portals_tools(mcp, make_request)
print(" ✓ portals (8 tools)")
register_work_history_tools(mcp, make_request)
print(" ✓ work_history (3 tools)")
register_tags_tools(mcp)
print(" ✓ tags (2 tools)")
register_webhooks_tools(mcp)
print(" ✓ webhooks (4 tools)")
register_users_tools(mcp)
print(" ✓ users (2 tools)")
register_triggers_tools(mcp)
print(" ✓ triggers (2 tools)")
register_attachments_tools(mcp)
print(" ✓ attachments (4 tools)")
register_backups_tools(mcp)
print(" ✓ backups (3 tools)")
register_events_tools(mcp)
print(" ✓ events (5 tools)")
# Register prompts, resources, and templates
print("\nLoading prompts, resources, and templates...")
register_recruiting_prompts(mcp, make_request)
print(" ✓ recruiting prompts (3 prompts)")
register_interview_prompts(mcp, make_request)
print(" ✓ interview prompts (1 prompt)")
register_candidate_resources(mcp, make_request)
print(" ✓ candidate resources (2 resources)")
register_job_resources(mcp, make_request)
print(" ✓ job resources (2 resources)")
register_email_templates(mcp)
print(" ✓ email templates (5 templates)")
print("\n✅ All loaded:")
print(" - 164 tools")
print(" - 4 prompts")
print(" - 4 resources")
print(" - 5 templates\n")
if __name__ == "__main__":
# Just run the server - tools already registered
mcp.run(transport="stdio")