Merge pull request #57 from pamenon/main

Added python mcp project files

Author: erouvas

python-mcp-sample/.gitignore

@@ -0,0 +1,5 @@
+*.db
+__pycache__/
+.venv/
+*.pyc
+.python-version

python-mcp-sample/mcp_server.py

@@ -0,0 +1,120 @@
+"""MCP server exposing insurance policy data from SQLite."""
+
+import os
+import sqlite3
+from textwrap import dedent
+
+from mcp.server.fastmcp import FastMCP
+
+DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "insurance.db")
+
+mcp = FastMCP("Insurance Policy Server")
+
+
+def _get_connection() -> sqlite3.Connection:
+    conn = sqlite3.connect(DB_PATH)
+    conn.row_factory = sqlite3.Row
+    return conn
+
+
+def _format_policy(row: sqlite3.Row) -> str:
+    return dedent(f"""\
+        Policy Number : {row["policy_number"]}
+        Holder        : {row["holder_name"]} ({row["holder_email"]})
+        Type          : {row["policy_type"]}
+        Premium       : ${row["premium"]:,.2f}
+        Deductible    : ${row["deductible"]:,.2f}
+        Coverage Limit: ${row["coverage_limit"]:,.2f}
+        Period        : {row["start_date"]} to {row["end_date"]}
+        Status        : {row["status"]}""")
+
+
+@mcp.tool()
+def get_policy(policy_number: str) -> str:
+    """Look up a single insurance policy by its policy number.
+
+    Args:
+        policy_number: The unique policy identifier (e.g. POL-2024005).
+    """
+    conn = _get_connection()
+    row = conn.execute(
+        "SELECT * FROM policies WHERE policy_number = ?", (policy_number,)
+    ).fetchone()
+    conn.close()
+
+    if row is None:
+        return f"No policy found with number '{policy_number}'."
+    return _format_policy(row)
+
+
+@mcp.tool()
+def list_policies(status: str | None = None, policy_type: str | None = None) -> str:
+    """List insurance policies with optional filters.
+
+    Args:
+        status: Filter by status (Active, Expired, or Cancelled). None returns all.
+        policy_type: Filter by type (Auto, Home, Life, or Health). None returns all.
+    """
+    query = "SELECT * FROM policies WHERE 1=1"
+    params: list[str] = []
+
+    if status is not None:
+        query += " AND status = ?"
+        params.append(status)
+    if policy_type is not None:
+        query += " AND policy_type = ?"
+        params.append(policy_type)
+
+    query += " ORDER BY policy_number"
+
+    conn = _get_connection()
+    rows = conn.execute(query, params).fetchall()
+    conn.close()
+
+    if not rows:
+        return "No policies match the given filters."
+
+    header = f"{'#':<4} {'Policy Number':<16} {'Holder':<25} {'Type':<8} {'Premium':>10} {'Status':<10}"
+    separator = "-" * len(header)
+    lines = [header, separator]
+
+    for i, row in enumerate(rows, 1):
+        lines.append(
+            f"{i:<4} {row['policy_number']:<16} {row['holder_name']:<25} "
+            f"{row['policy_type']:<8} ${row['premium']:>9,.2f} {row['status']:<10}"
+        )
+
+    lines.append(separator)
+    lines.append(f"Total: {len(rows)} policies")
+    return "\n".join(lines)
+
+
+@mcp.tool()
+def send_policy_to_agent(policy_number: str) -> str:
+    """Retrieve an insurance policy and send its data to the AI agent for processing.
+
+    Args:
+        policy_number: The unique policy identifier to look up and forward.
+    """
+    conn = _get_connection()
+    row = conn.execute(
+        "SELECT * FROM policies WHERE policy_number = ?", (policy_number,)
+    ).fetchone()
+    conn.close()
+
+    if row is None:
+        return f"No policy found with number '{policy_number}'. Nothing sent."
+
+    policy_data = dict(row)
+
+    # TODO: send extra info to the agent
+    print(f"TODO: send policy {policy_number} data to AI agent: {policy_data}")
+
+    return (
+        f"Policy {policy_number} data retrieved and queued for the AI agent (not yet implemented).\n\n"
+        + _format_policy(row)
+    )
+
+
+if __name__ == "__main__":
+    mcp.run()

python-mcp-sample/pyproject.toml

@@ -0,0 +1,9 @@
+[project]
+name = "mcp-insurance"
+version = "0.1.0"
+description = "MCP server for insurance policy data"
+requires-python = ">=3.10"
+dependencies = [
+    "faker>=40.11.0",
+    "mcp[cli]>=1.26.0",
+]

python-mcp-sample/setup_db.py

@@ -0,0 +1,143 @@
+"""Create and populate the insurance.db SQLite database with fake policy data."""
+
+import os
+import random
+import sqlite3
+from datetime import timedelta
+
+from faker import Faker
+
+DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "insurance.db")
+NUM_POLICIES = 30
+
+POLICY_TYPES = ["Auto", "Home", "Life", "Health"]
+STATUSES = ["Active", "Expired", "Cancelled"]
+STATUS_WEIGHTS = [0.6, 0.25, 0.15]
+
+PREMIUM_RANGES = {
+    "Auto": (600, 3000),
+    "Home": (800, 4000),
+    "Life": (300, 2000),
+    "Health": (1500, 8000),
+}
+
+DEDUCTIBLE_RANGES = {
+    "Auto": (250, 2000),
+    "Home": (500, 5000),
+    "Life": (0, 500),
+    "Health": (500, 6000),
+}
+
+COVERAGE_LIMIT_RANGES = {
+    "Auto": (25_000, 300_000),
+    "Home": (100_000, 1_000_000),
+    "Life": (50_000, 2_000_000),
+    "Health": (100_000, 5_000_000),
+}
+
+CREATE_TABLE_SQL = """
+CREATE TABLE IF NOT EXISTS policies (
+    id              INTEGER PRIMARY KEY AUTOINCREMENT,
+    policy_number   TEXT UNIQUE NOT NULL,
+    holder_name     TEXT NOT NULL,
+    holder_email    TEXT NOT NULL,
+    policy_type     TEXT NOT NULL,
+    premium         REAL NOT NULL,
+    deductible      REAL NOT NULL,
+    coverage_limit  REAL NOT NULL,
+    start_date      TEXT NOT NULL,
+    end_date        TEXT NOT NULL,
+    status          TEXT NOT NULL
+);
+"""
+
+INSERT_SQL = """
+INSERT INTO policies (
+    policy_number, holder_name, holder_email, policy_type,
+    premium, deductible, coverage_limit, start_date, end_date, status
+) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?);
+"""
+
+
+def generate_policy_number(fake: Faker, index: int) -> str:
+    prefix = fake.random_element(["POL", "INS", "COV"])
+    return f"{prefix}-{2024_000 + index:07d}"
+
+
+def build_policy(fake: Faker, index: int) -> tuple:
+    policy_number = generate_policy_number(fake, index)
+    holder_name = fake.name()
+    holder_email = fake.email()
+    policy_type = fake.random_element(POLICY_TYPES)
+    status = random.choices(STATUSES, weights=STATUS_WEIGHTS, k=1)[0]
+
+    low, high = PREMIUM_RANGES[policy_type]
+    premium = round(random.uniform(low, high), 2)
+
+    low, high = DEDUCTIBLE_RANGES[policy_type]
+    deductible = round(random.uniform(low, high), 2)
+
+    low, high = COVERAGE_LIMIT_RANGES[policy_type]
+    coverage_limit = round(random.uniform(low, high), 2)
+
+    start_date = fake.date_between(start_date="-3y", end_date="today")
+    end_date = start_date + timedelta(days=365)
+
+    return (
+        policy_number,
+        holder_name,
+        holder_email,
+        policy_type,
+        premium,
+        deductible,
+        coverage_limit,
+        start_date.isoformat(),
+        end_date.isoformat(),
+        status,
+    )
+
+
+def main() -> None:
+    if os.path.exists(DB_PATH):
+        os.remove(DB_PATH)
+        print(f"Removed existing database: {DB_PATH}")
+
+    conn = sqlite3.connect(DB_PATH)
+    cursor = conn.cursor()
+    cursor.execute(CREATE_TABLE_SQL)
+
+    fake = Faker()
+    Faker.seed(42)
+    random.seed(42)
+
+    policies = [build_policy(fake, i) for i in range(NUM_POLICIES)]
+    cursor.executemany(INSERT_SQL, policies)
+    conn.commit()
+
+    cursor.execute("SELECT COUNT(*) FROM policies")
+    count = cursor.fetchone()[0]
+
+    cursor.execute(
+        "SELECT policy_type, COUNT(*) FROM policies GROUP BY policy_type ORDER BY policy_type"
+    )
+    breakdown = cursor.fetchall()
+
+    cursor.execute(
+        "SELECT status, COUNT(*) FROM policies GROUP BY status ORDER BY status"
+    )
+    status_breakdown = cursor.fetchall()
+
+    conn.close()
+
+    print(f"\nDatabase created at: {DB_PATH}")
+    print(f"Total policies inserted: {count}\n")
+    print("By type:")
+    for ptype, cnt in breakdown:
+        print(f"  {ptype:<10} {cnt}")
+    print("\nBy status:")
+    for status, cnt in status_breakdown:
+        print(f"  {status:<12} {cnt}")
+
+
+if __name__ == "__main__":
+    main()