KYC MCP Server

""" Interactive testing script for KYC tools. This script provides an interactive CLI for manually testing the KYC MCP server tools. """ import asyncio import sys from typing import Optional import httpx class Colors: """ANSI color codes for terminal output.""" HEADER = '\033[95m' BLUE = '\033[94m' CYAN = '\033[96m' GREEN = '\033[92m' YELLOW = '\033[93m' RED = '\033[91m' END = '\033[0m' BOLD = '\033[1m' class KYCTester: """Interactive tester for KYC tools.""" def __init__(self, server_url: str = "http://localhost:8000"): """Initialize tester with server URL.""" self.server_url = server_url self.client: Optional[httpx.AsyncClient] = None async def connect(self): """Connect to MCP server.""" self.client = httpx.AsyncClient(base_url=self.server_url, timeout=30.0) try: response = await self.client.get("/health") if response.status_code == 200: print(f"{Colors.GREEN}✓ Connected to server at {self.server_url}{Colors.END}\n") return True except Exception as e: print(f"{Colors.RED}✗ Failed to connect: {e}{Colors.END}\n") return False async def disconnect(self): """Disconnect from server.""" if self.client: await self.client.aclose() def print_header(self, text: str): """Print section header.""" print(f"\n{Colors.BOLD}{Colors.CYAN}{'='*60}{Colors.END}") print(f"{Colors.BOLD}{Colors.CYAN}{text}{Colors.END}") print(f"{Colors.BOLD}{Colors.CYAN}{'='*60}{Colors.END}\n") def print_success(self, text: str): """Print success message.""" print(f"{Colors.GREEN}✓ {text}{Colors.END}") def print_error(self, text: str): """Print error message.""" print(f"{Colors.RED}✗ {text}{Colors.END}") def print_info(self, text: str): """Print info message.""" print(f"{Colors.BLUE}ℹ {text}{Colors.END}") def get_input(self, prompt: str, default: str = "") -> str: """Get user input with optional default.""" if default: prompt = f"{prompt} [{default}]: " else: prompt = f"{prompt}: " value = input(f"{Colors.YELLOW}{prompt}{Colors.END}").strip() return value if value else default async def test_pan_verification(self): """Interactive PAN verification test.""" self.print_header("PAN Verification Test") # Get input pan = self.get_input("Enter PAN number", "ABCDE1234F") name = self.get_input("Enter name as per PAN", "John Doe") dob = self.get_input("Enter date of birth (DD/MM/YYYY)", "01/01/1990") consent = self.get_input("Enter consent (Y/y)", "Y") reason = self.get_input("Enter reason", "Testing") # Prepare request payload = { "tool": "verify_pan", "params": { "pan": pan, "name_as_per_pan": name, "date_of_birth": dob, "consent": consent, "reason": reason } } print(f"\n{Colors.BLUE}→ Sending request...{Colors.END}") try: response = await self.client.post("/tools/execute", json=payload) if response.status_code == 200: result = response.json() self.print_success("PAN verification successful!") print(f"\n{Colors.BOLD}Results:{Colors.END}") print(f" PAN: {result.get('pan')}") print(f" Category: {result.get('category')}") print(f" Status: {result.get('status')}") print(f" Name Match: {Colors.GREEN if result.get('name_match') else Colors.RED}{result.get('name_match')}{Colors.END}") print(f" DOB Match: {Colors.GREEN if result.get('dob_match') else Colors.RED}{result.get('dob_match')}{Colors.END}") print(f" Aadhaar Seeding: {result.get('aadhaar_seeding_status')}") if result.get('remarks'): print(f" Remarks: {Colors.YELLOW}{result.get('remarks')}{Colors.END}") else: self.print_error(f"Request failed with status {response.status_code}") print(f"Response: {response.text}") except Exception as e: self.print_error(f"Error: {e}") async def test_pan_aadhaar_link(self): """Interactive PAN-Aadhaar link test.""" self.print_header("PAN-Aadhaar Link Status Test") # Get input pan = self.get_input("Enter individual PAN (4th char must be 'P')", "ABCPE1234F") aadhaar = self.get_input("Enter 12-digit Aadhaar number", "123456789012") consent = self.get_input("Enter consent (Y/y)", "Y") reason = self.get_input("Enter reason", "Testing") # Prepare request payload = { "tool": "check_pan_aadhaar_link", "params": { "pan": pan, "aadhaar_number": aadhaar, "consent": consent, "reason": reason } } print(f"\n{Colors.BLUE}→ Sending request...{Colors.END}") try: response = await self.client.post("/tools/execute", json=payload) if response.status_code == 200: result = response.json() self.print_success("Link status check successful!") print(f"\n{Colors.BOLD}Results:{Colors.END}") linked = result.get('linked') status_color = Colors.GREEN if linked else Colors.YELLOW print(f" Linked: {status_color}{linked}{Colors.END}") print(f" Status: {result.get('status')}") print(f" Message: {result.get('message')}") else: self.print_error(f"Request failed with status {response.status_code}") print(f"Response: {response.text}") except Exception as e: self.print_error(f"Error: {e}") async def list_tools(self): """List available tools.""" self.print_header("Available Tools") try: response = await self.client.get("/tools") if response.status_code == 200: tools = response.json() self.print_success(f"Found {len(tools)} tools") for i, tool in enumerate(tools, 1): print(f"\n{Colors.BOLD}{i}. {tool.get('name')}{Colors.END}") print(f" {tool.get('description')}") else: self.print_error(f"Failed to list tools: {response.status_code}") except Exception as e: self.print_error(f"Error: {e}") async def run_menu(self): """Run interactive menu.""" while True: self.print_header("KYC Tools Interactive Tester") print("Select an option:") print(f" {Colors.BOLD}1.{Colors.END} Test PAN Verification") print(f" {Colors.BOLD}2.{Colors.END} Test PAN-Aadhaar Link Status") print(f" {Colors.BOLD}3.{Colors.END} List Available Tools") print(f" {Colors.BOLD}4.{Colors.END} Exit") choice = self.get_input("\nEnter choice (1-4)", "1") if choice == "1": await self.test_pan_verification() elif choice == "2": await self.test_pan_aadhaar_link() elif choice == "3": await self.list_tools() elif choice == "4": print(f"\n{Colors.GREEN}Goodbye!{Colors.END}\n") break else: self.print_error("Invalid choice. Please enter 1-4.") # Ask to continue if choice in ["1", "2", "3"]: input(f"\n{Colors.YELLOW}Press Enter to continue...{Colors.END}") async def main(): """Main entry point.""" print(f"\n{Colors.BOLD}{Colors.HEADER}{'='*60}{Colors.END}") print(f"{Colors.BOLD}{Colors.HEADER}KYC MCP Server - Interactive Testing Tool{Colors.END}") print(f"{Colors.BOLD}{Colors.HEADER}{'='*60}{Colors.END}\n") # Get server URL if len(sys.argv) > 1: server_url = sys.argv[1] else: server_url = input(f"{Colors.YELLOW}Enter server URL [http://localhost:8000]: {Colors.END}").strip() if not server_url: server_url = "http://localhost:8000" # Create tester tester = KYCTester(server_url) # Connect to server if await tester.connect(): try: await tester.run_menu() finally: await tester.disconnect() else: print(f"{Colors.RED}Failed to connect to server. Please check if the server is running.{Colors.END}\n") sys.exit(1) if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: print(f"\n\n{Colors.YELLOW}Interrupted by user{Colors.END}\n") sys.exit(0)