feat(mcp): Add MCP server for Corrigo SDK via FastMCP 3.0

pyproject.toml

@@ -8,7 +8,7 @@ version = "0.1.0"
 description = "Python SDK for the Corrigo Enterprise REST API"
 readme = "README.md"
 license = "MIT"
-requires-python = ">=3.9"
+requires-python = ">=3.10"
 authors = [
     { name = "Spencer Bean" }
 ]
@@ -19,7 +19,6 @@ classifiers = [
     "License :: OSI Approved :: MIT License",
     "Operating System :: OS Independent",
     "Programming Language :: Python :: 3",
-    "Programming Language :: Python :: 3.9",
     "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",
@@ -34,8 +33,12 @@ dependencies = [
 
 [project.scripts]
 corrigo = "corrigo.cli:main"
+corrigo-mcp = "corrigo.mcp:main"
 
 [project.optional-dependencies]
+mcp = [
+    "fastmcp>=3.0.0",
+]
 cli = [
     "typer>=0.9.0",
     "rich>=13.0.0",
@@ -71,7 +74,7 @@ include = [
 packages = ["src/corrigo"]
 
 [tool.ruff]
-target-version = "py39"
+target-version = "py310"
 line-length = 100
 
 [tool.ruff.lint]
@@ -95,7 +98,7 @@ ignore = [
 known-first-party = ["corrigo"]
 
 [tool.mypy]
-python_version = "3.9"
+python_version = "3.10"
 strict = true
 warn_return_any = true
 warn_unused_ignores = true

src/corrigo/mcp/__init__.py

@@ -0,0 +1,15 @@
+"""Corrigo MCP Server — exposes the Corrigo SDK via Model Context Protocol."""
+
+try:
+    from corrigo.mcp.server import mcp
+except ImportError:
+    raise ImportError(
+        "The MCP server requires FastMCP. Install with: pip install corrigo[mcp]"
+    ) from None
+
+__all__ = ["main", "mcp"]
+
+
+def main() -> None:
+    """Entry point for the corrigo-mcp CLI command."""
+    mcp.run()

src/corrigo/mcp/__main__.py

@@ -0,0 +1,5 @@
+"""Allow running the MCP server with: python -m corrigo.mcp"""
+
+from corrigo.mcp import main
+
+main()

src/corrigo/mcp/prompts/__init__.py

@@ -0,0 +1,11 @@
+"""Prompt registrations for the Corrigo MCP server.
+
+Importing this module registers all prompts on the server via side effects.
+"""
+
+from corrigo.mcp.prompts import diagnose as _diagnose  # noqa: F401
+from corrigo.mcp.prompts import intake as _intake  # noqa: F401
+from corrigo.mcp.prompts import overview as _overview  # noqa: F401
+from corrigo.mcp.prompts import status as _status  # noqa: F401
+from corrigo.mcp.prompts import triage as _triage  # noqa: F401
+from corrigo.mcp.prompts import troubleshoot as _troubleshoot  # noqa: F401

src/corrigo/mcp/prompts/diagnose.py

@@ -0,0 +1,66 @@
+"""Diagnose work order prompt for technicians/ops."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+
+from fastmcp import Context
+from fastmcp.prompts.prompt import Message
+
+from corrigo.mcp.server import mcp
+
+
+@mcp.prompt(tags={"internal", "work-orders"})
+async def diagnose_work_order(work_order_id: int, ctx: Context) -> list[Message]:
+    """Deep diagnostic context for a work order. Internal use only.
+
+    Fetches full technical context including equipment attributes, service
+    history, and customer details. Used by technicians and operations teams.
+    """
+    client = ctx.lifespan_context["client"]
+
+    await ctx.info(f"Fetching work order {work_order_id} for diagnosis...")
+    wo = await asyncio.to_thread(client.work_orders.get, work_order_id)
+
+    # Fetch asset with attributes
+    asset_info = "[Could not load asset data]"
+    try:
+        asset_id = None
+        items = wo.get("Items", [])
+        if items:
+            asset_id = items[0].get("Asset", {}).get("Id")
+        if asset_id:
+            asset = await asyncio.to_thread(client.locations.get_with_attributes, asset_id)
+            asset_info = json.dumps(asset, indent=2, default=str)
+    except Exception as e:
+        asset_info = f"[Could not load asset: {e}]"
+
+    # Fetch customer
+    customer_info = "[Could not load customer data]"
+    try:
+        customer_id = wo.get("Customer", {}).get("Id")
+        if customer_id:
+            customer = await asyncio.to_thread(client.customers.get, customer_id)
+            customer_info = json.dumps(customer, indent=2, default=str)
+    except Exception as e:
+        customer_info = f"[Could not load customer: {e}]"
+
+    return [
+        Message(
+            content=f"""Diagnose the following work order:
+
+Work Order: {json.dumps(wo, indent=2, default=str)}
+
+Equipment: {asset_info}
+
+Customer: {customer_info}
+
+Provide a technical diagnosis considering:
+1. Equipment make, model, and age (from attributes)
+2. Common failure modes for this equipment type
+3. Parts that may be needed
+4. Estimated repair complexity and time
+5. Safety considerations""",
+        )
+    ]

src/corrigo/mcp/prompts/intake.py

@@ -0,0 +1,60 @@
+"""Issue intake prompt for customer-facing use (call center/voice agent)."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+
+from fastmcp import Context
+from fastmcp.prompts.prompt import Message
+
+from corrigo.mcp.server import mcp
+
+
+@mcp.prompt(tags={"customer-facing", "work-orders"})
+async def report_issue(customer_id: int, description: str, ctx: Context) -> list[Message]:
+    """Guide intake of a new service request from a customer.
+
+    Fetches customer details and assets to help identify the right equipment
+    and collect information needed for work order creation.
+    """
+    client = ctx.lifespan_context["client"]
+
+    await ctx.info("Fetching customer and assets...")
+    customer = await asyncio.to_thread(client.customers.get, customer_id)
+
+    # Fetch customer assets with graceful degradation
+    equipment_info = "[Could not load equipment list]"
+    try:
+        assets = await asyncio.to_thread(client.locations.list_by_customer, customer_id)
+        equipment = [
+            {"id": a["Id"], "name": a.get("Name", "Unknown")}
+            for a in assets
+            if a.get("TypeId") == "Equipment"
+        ]
+        if equipment:
+            equipment_info = json.dumps(equipment, indent=2)
+        else:
+            equipment_info = "No equipment found for this customer."
+    except Exception as e:
+        equipment_info = f"[Could not load equipment: {e}]"
+
+    return [
+        Message(
+            role="user",
+            content=f"""A customer is reporting an issue.
+
+Customer: {customer.get("DisplayAs", "Unknown")} (ID: {customer_id})
+Issue: {description}
+
+Equipment at this location:
+{equipment_info}
+
+Help identify which equipment is affected, confirm the issue details,
+and if a work order is needed, call the create_work_order tool with:
+- customer_id: {customer_id}
+- asset_id: [the identified equipment ID]
+- task_id: [appropriate task ID]
+- subtype_id: [appropriate subtype ID]""",
+        )
+    ]

src/corrigo/mcp/prompts/overview.py

@@ -0,0 +1,61 @@
+"""Customer overview prompt for account managers."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+
+from fastmcp import Context
+from fastmcp.prompts.prompt import Message
+
+from corrigo.mcp.server import mcp
+
+
+@mcp.prompt(tags={"internal", "customers"})
+async def customer_overview(customer_id: int, ctx: Context) -> list[Message]:
+    """Generate an overview of a customer for account management.
+
+    Fetches customer details, assets, and recent work orders to provide
+    a comprehensive view of the customer's facilities and service history.
+    """
+    client = ctx.lifespan_context["client"]
+
+    await ctx.info(f"Fetching customer {customer_id} overview...")
+    customer = await asyncio.to_thread(client.customers.get, customer_id)
+
+    # Fetch assets
+    assets_info = "[Could not load assets]"
+    try:
+        assets = await asyncio.to_thread(client.locations.list_by_customer, customer_id)
+        assets_info = json.dumps(assets, indent=2, default=str)
+    except Exception as e:
+        assets_info = f"[Could not load assets: {e}]"
+
+    # Fetch recent work orders
+    wo_info = "[Could not load work orders]"
+    try:
+        work_orders = await asyncio.to_thread(
+            client.work_orders.list_by_customer, customer_id, limit=20
+        )
+        wo_info = json.dumps(work_orders, indent=2, default=str)
+    except Exception as e:
+        wo_info = f"[Could not load work orders: {e}]"
+
+    return [
+        Message(
+            role="user",
+            content=f"""Provide an overview of this customer:
+
+Customer: {json.dumps(customer, indent=2, default=str)}
+
+Assets/Locations: {assets_info}
+
+Recent Work Orders: {wo_info}
+
+Summarize:
+1. Customer profile and key details
+2. Number and types of assets
+3. Work order trends (frequency, types, common issues)
+4. Any concerns or opportunities for the account""",
+        )
+    ]

src/corrigo/mcp/prompts/status.py

@@ -0,0 +1,47 @@
+"""Work order status prompt for customer-facing use."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+
+from fastmcp import Context
+from fastmcp.prompts.prompt import Message
+
+from corrigo.mcp.server import mcp
+
+
+@mcp.prompt(tags={"customer-facing", "work-orders"})
+async def work_order_status(work_order_id: int, ctx: Context) -> list[Message]:
+    """Provide customer-safe work order status information.
+
+    Returns only information appropriate for sharing with customers:
+    status, last update, and next steps. Excludes internal notes,
+    cost data, and technician details.
+    """
+    client = ctx.lifespan_context["client"]
+
+    await ctx.info(f"Fetching work order {work_order_id} status...")
+    wo = await asyncio.to_thread(client.work_orders.get, work_order_id)
+
+    # Extract only customer-safe fields
+    safe_fields = {
+        "Number": wo.get("Number"),
+        "Status": wo.get("StatusId"),
+        "Created": wo.get("DtCreated"),
+        "LastUpdated": wo.get("DtModified"),
+        "TypeCategory": wo.get("TypeCategory"),
+    }
+
+    return [
+        Message(
+            role="user",
+            content=f"""Provide a customer-friendly status update for this work order:
+
+{json.dumps(safe_fields, indent=2, default=str)}
+
+Provide a brief, friendly status update suitable for sharing with the customer.
+Do NOT mention internal details, costs, or technician assignments.
+Focus on: current status, what has been done, and expected next steps.""",
+        )
+    ]

src/corrigo/mcp/prompts/triage.py

@@ -0,0 +1,43 @@
+"""Triage work order prompt for internal ops/dispatch."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+
+from fastmcp import Context
+
+from corrigo.mcp.server import mcp
+
+
+@mcp.prompt(tags={"internal", "work-orders"})
+async def triage_work_order(work_order_id: int, ctx: Context) -> str:
+    """Triage a work order for dispatch. Fetches full context for prioritization.
+
+    Internal prompt for ops/dispatch teams. Provides the work order details,
+    customer info, and asset data to help decide priority and assignment.
+    """
+    client = ctx.lifespan_context["client"]
+
+    await ctx.info(f"Fetching work order {work_order_id} for triage...")
+    wo = await asyncio.to_thread(client.work_orders.get, work_order_id)
+
+    # Fetch related data with graceful degradation
+    customer_info = "[Could not load customer data]"
+    try:
+        customer_id = wo.get("Customer", {}).get("Id")
+        if customer_id:
+            customer = await asyncio.to_thread(client.customers.get, customer_id)
+            customer_info = json.dumps(customer, indent=2, default=str)
+    except Exception as e:
+        customer_info = f"[Could not load customer: {e}]"
+
+    return f"""Triage the following work order for dispatch:
+
+Work Order: {json.dumps(wo, indent=2, default=str)}
+
+Customer: {customer_info}
+
+Assess priority, recommend assignment, and identify any urgency factors.
+Consider: equipment type, customer SLA tier, time since creation, and
+whether this is a repeat issue."""