#!/usr/bin/env python3 """ data_monitor_cli.py -- Data platform monitoring CLI agent. Uses the Acme SaaS MCP server (notebooks/mcp_data_server.py) and Claude to answer monitoring questions from the terminal. Demonstrates the D2 pattern: a thin CLI application backed by a reusable MCP server. Usage examples: # Ask a direct question python scripts/data_monitor_cli.py "Which warehouse has the highest cost this week?" # Use a server-side prompt template python scripts/data_monitor_cli.py --prompt incident_analysis \\ --warehouse WH_BI_M --date-range "2025-07-20 to 2025-08-10" # List available tools, resources, and prompts python scripts/data_monitor_cli.py --list-tools Requirements: - ANTHROPIC_API_KEY in .env or environment - mcp>=1.0.0, anthropic>=0.40.0, nest_asyncio>=1.5.0 - Datasets in data/ (run: python scripts/generate_data.py) """ from __future__ import annotations import argparse import asyncio import os import sys from pathlib import Path import anthropic import nest_asyncio from dotenv import load_dotenv from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client # Patch the event loop for nest_asyncio compatibility nest_asyncio.apply() # -------------------------------------------------------------------------- # Configuration # -------------------------------------------------------------------------- _HERE = Path(__file__).resolve().parent # scripts/ _ROOT = _HERE.parent # project root SERVER_SCRIPT = _ROOT / "notebooks" / "mcp_data_server.py" SERVER_PARAMS = StdioServerParameters( command=sys.executable, args=[str(SERVER_SCRIPT)], ) MODEL = "claude-sonnet-4-5" # -------------------------------------------------------------------------- # Helper: run coroutine from sync context # -------------------------------------------------------------------------- def run(coro): return asyncio.get_event_loop().run_until_complete(coro) # -------------------------------------------------------------------------- # Core agent loop # -------------------------------------------------------------------------- async def run_agent(question: str, verbose: bool = False) -> str: """Connect to MCP server, run Claude with discovered tools, return final answer.""" client = anthropic.Anthropic() async with stdio_client(SERVER_PARAMS) as (read, write): async with ClientSession(read, write) as session: await session.initialize() # Discover tools dynamically -- no hardcoded schemas here tools_result = await session.list_tools() anthropic_tools = [ { "name": t.name, "description": t.description or "", "input_schema": t.inputSchema, } for t in tools_result.tools ] if verbose: print( f"[agent] Connected to {session.server_info.name}. " f"{len(anthropic_tools)} tools available.", file=sys.stderr, ) messages = [{"role": "user", "content": question}] for turn in range(1, 11): response = client.messages.create( model=MODEL, max_tokens=4096, tools=anthropic_tools, messages=messages, ) if verbose: tool_calls = [b.name for b in response.content if b.type == "tool_use"] print( f"[agent] Turn {turn}: stop_reason={response.stop_reason}" + (f", tools={tool_calls}" if tool_calls else ""), file=sys.stderr, ) if response.stop_reason == "end_turn": return next( (b.text for b in response.content if b.type == "text"), "(no text response)", ) if response.stop_reason == "tool_use": messages.append({"role": "assistant", "content": response.content}) tool_results = [] for block in response.content: if block.type == "tool_use": mcp_result = await session.call_tool(block.name, block.input) tool_results.append({ "type": "tool_result", "tool_use_id": block.id, "content": mcp_result.content[0].text, }) messages.append({"role": "user", "content": tool_results}) return "(max turns reached -- no end_turn response)" # -------------------------------------------------------------------------- # Utility: list tools and resources # -------------------------------------------------------------------------- async def list_server_capabilities() -> None: async with stdio_client(SERVER_PARAMS) as (read, write): async with ClientSession(read, write) as session: await session.initialize() print(f"Server: {session.server_info.name} v{session.server_info.version}") print() tools = await session.list_tools() print(f"Tools ({len(tools.tools)}):") for t in tools.tools: print(f" {t.name}") if t.description: first_line = t.description.strip().splitlines()[0] print(f" {first_line}") print() resources = await session.list_resources() print(f"Resources ({len(resources.resources)}):") for r in resources.resources: print(f" {r.uri} -- {r.name}") print() prompts = await session.list_prompts() print(f"Prompts ({len(prompts.prompts)}):") for p in prompts.prompts: print(f" {p.name}: {p.description or ''}") if p.arguments: for arg in p.arguments: req = " (required)" if arg.required else " (optional)" print(f" --{arg.name}{req}") async def get_prompt_text(name: str, warehouse: str, date_range: str) -> str: async with stdio_client(SERVER_PARAMS) as (read, write): async with ClientSession(read, write) as session: await session.initialize() result = await session.get_prompt( name, arguments={"warehouse_name": warehouse, "date_range": date_range}, ) return result.messages[0].content.text # -------------------------------------------------------------------------- # Entry point # -------------------------------------------------------------------------- def main() -> None: load_dotenv(_ROOT / ".env") if not os.environ.get("ANTHROPIC_API_KEY"): print("ERROR: ANTHROPIC_API_KEY not set. Add it to .env or export it.", file=sys.stderr) sys.exit(1) if not SERVER_SCRIPT.exists(): print(f"ERROR: MCP server not found at {SERVER_SCRIPT}", file=sys.stderr) sys.exit(1) parser = argparse.ArgumentParser( description="Data platform monitoring CLI -- powered by Claude + MCP", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__, ) parser.add_argument( "question", nargs="?", help="Natural language question for the monitoring agent", ) parser.add_argument( "--prompt", metavar="NAME", help="Use a server-side prompt template (e.g. incident_analysis)", ) parser.add_argument( "--warehouse", default="WH_BI_M", help="Warehouse name for prompt templates (default: WH_BI_M)", ) parser.add_argument( "--date-range", default="2025-07-01 to 2025-09-28", help='Date range for prompt templates (default: "2025-07-01 to 2025-09-28")', ) parser.add_argument( "--list-tools", action="store_true", help="List available tools, resources, and prompts, then exit", ) parser.add_argument( "-v", "--verbose", action="store_true", help="Show agent turn-by-turn progress on stderr", ) args = parser.parse_args() # --list-tools: enumerate server capabilities and exit if args.list_tools: run(list_server_capabilities()) return # Determine the question to ask if args.prompt: question = run(get_prompt_text(args.prompt, args.warehouse, args.date_range)) if args.verbose: print(f"[agent] Using prompt '{args.prompt}' (rendered).", file=sys.stderr) elif args.question: question = args.question else: parser.print_help() sys.exit(1) # Run the agent if args.verbose: print(f"[agent] Question: {question[:120]}...", file=sys.stderr) answer = run(run_agent(question, verbose=args.verbose)) print(answer) if __name__ == "__main__": main()