mcp_client_example.py•11.1 kB
"""
Example MCP client for connecting to and using the KYC MCP Server.
This example demonstrates:
- Connecting to the MCP server
- Listing available tools
- Calling tools with parameters
- Handling responses and errors
"""
import asyncio
import json
from typing import Any, Dict, List
class MCPClient:
"""Simple MCP client for demonstration purposes."""
def __init__(self, server_url: str = "http://localhost:8000"):
"""
Initialize MCP client.
Args:
server_url: URL of the MCP server
"""
self.server_url = server_url
self.session = None
async def connect(self):
"""Establish connection to MCP server."""
import httpx
self.session = httpx.AsyncClient(base_url=self.server_url, timeout=30.0)
print(f"✓ Connected to MCP server at {self.server_url}")
async def disconnect(self):
"""Close connection to MCP server."""
if self.session:
await self.session.aclose()
print("✓ Disconnected from MCP server")
async def list_tools(self) -> List[Dict[str, Any]]:
"""
List all available tools on the server.
Returns:
List of tool definitions
"""
response = await self.session.get("/tools")
response.raise_for_status()
tools = response.json()
print(f"\n✓ Found {len(tools)} available tools")
return tools
async def call_tool(self, tool_name: str, params: Dict[str, Any]) -> Dict[str, Any]:
"""
Call a tool with given parameters.
Args:
tool_name: Name of the tool to call
params: Parameters for the tool
Returns:
Tool execution result
"""
payload = {
"tool": tool_name,
"params": params
}
print(f"\n→ Calling tool: {tool_name}")
print(f" Parameters: {json.dumps(params, indent=2)}")
response = await self.session.post("/tools/execute", json=payload)
response.raise_for_status()
result = response.json()
print(f"✓ Tool executed successfully")
return result
async def get_tool_info(self, tool_name: str) -> Dict[str, Any]:
"""
Get detailed information about a specific tool.
Args:
tool_name: Name of the tool
Returns:
Tool metadata
"""
response = await self.session.get(f"/tools/{tool_name}")
response.raise_for_status()
return response.json()
async def example_pan_verification():
"""Example: Verify a PAN card."""
print("\n" + "="*60)
print("Example 1: PAN Verification")
print("="*60)
client = MCPClient()
try:
# Connect to server
await client.connect()
# Call PAN verification tool
result = await client.call_tool(
tool_name="verify_pan",
params={
"pan": "ABCDE1234F",
"name_as_per_pan": "John Doe",
"date_of_birth": "01/01/1990",
"consent": "Y",
"reason": "Account opening"
}
)
# Display results
print("\nVerification Results:")
print(f" PAN: {result.get('pan')}")
print(f" Status: {result.get('status')}")
print(f" Name Match: {result.get('name_match')}")
print(f" DOB Match: {result.get('dob_match')}")
print(f" Aadhaar Seeding: {result.get('aadhaar_seeding_status')}")
if result.get('remarks'):
print(f" Remarks: {result.get('remarks')}")
except Exception as e:
print(f"\n✗ Error: {e}")
finally:
await client.disconnect()
async def example_pan_aadhaar_link():
"""Example: Check PAN-Aadhaar link status."""
print("\n" + "="*60)
print("Example 2: PAN-Aadhaar Link Check")
print("="*60)
client = MCPClient()
try:
# Connect to server
await client.connect()
# Call PAN-Aadhaar link check tool
result = await client.call_tool(
tool_name="check_pan_aadhaar_link",
params={
"pan": "ABCPE1234F",
"aadhaar_number": "123456789012",
"consent": "Y",
"reason": "Link status verification"
}
)
# Display results
print("\nLink Status Results:")
print(f" Linked: {result.get('linked')}")
print(f" Status: {result.get('status')}")
print(f" Message: {result.get('message')}")
except Exception as e:
print(f"\n✗ Error: {e}")
finally:
await client.disconnect()
async def example_list_tools():
"""Example: List all available tools."""
print("\n" + "="*60)
print("Example 3: List Available Tools")
print("="*60)
client = MCPClient()
try:
# Connect to server
await client.connect()
# List all tools
tools = await client.list_tools()
# Display tool information
for i, tool in enumerate(tools, 1):
print(f"\n{i}. {tool.get('name')}")
print(f" Description: {tool.get('description')}")
if 'inputSchema' in tool:
schema = tool['inputSchema']
if 'properties' in schema:
print(f" Parameters:")
for param, details in schema['properties'].items():
required = param in schema.get('required', [])
req_str = " (required)" if required else ""
print(f" - {param}{req_str}: {details.get('type')}")
except Exception as e:
print(f"\n✗ Error: {e}")
finally:
await client.disconnect()
async def example_error_handling():
"""Example: Handling errors and validation."""
print("\n" + "="*60)
print("Example 4: Error Handling")
print("="*60)
client = MCPClient()
try:
await client.connect()
# Example 1: Invalid PAN format
print("\n→ Testing with invalid PAN format...")
try:
await client.call_tool(
tool_name="verify_pan",
params={
"pan": "INVALID", # Invalid format
"name_as_per_pan": "John Doe",
"date_of_birth": "01/01/1990",
"consent": "Y",
"reason": "Test"
}
)
except Exception as e:
print(f"✓ Caught expected error: {type(e).__name__}")
print(f" Message: {str(e)}")
# Example 2: Missing required field
print("\n→ Testing with missing required field...")
try:
await client.call_tool(
tool_name="verify_pan",
params={
"pan": "ABCDE1234F",
"name_as_per_pan": "John Doe",
# Missing date_of_birth
"consent": "Y",
"reason": "Test"
}
)
except Exception as e:
print(f"✓ Caught expected error: {type(e).__name__}")
print(f" Message: {str(e)}")
# Example 3: Invalid consent
print("\n→ Testing with invalid consent...")
try:
await client.call_tool(
tool_name="verify_pan",
params={
"pan": "ABCDE1234F",
"name_as_per_pan": "John Doe",
"date_of_birth": "01/01/1990",
"consent": "N", # Invalid consent
"reason": "Test"
}
)
except Exception as e:
print(f"✓ Caught expected error: {type(e).__name__}")
print(f" Message: {str(e)}")
except Exception as e:
print(f"\n✗ Unexpected error: {e}")
finally:
await client.disconnect()
async def example_batch_operations():
"""Example: Batch processing multiple requests."""
print("\n" + "="*60)
print("Example 5: Batch Operations")
print("="*60)
client = MCPClient()
try:
await client.connect()
# Batch of PAN verifications
pans_to_verify = [
{
"pan": "ABCDE1234F",
"name_as_per_pan": "John Doe",
"date_of_birth": "01/01/1990"
},
{
"pan": "XYZPQ5678R",
"name_as_per_pan": "Jane Smith",
"date_of_birth": "15/06/1985"
},
{
"pan": "LMNOP9012S",
"name_as_per_pan": "Bob Johnson",
"date_of_birth": "31/12/2000"
}
]
print(f"\n→ Processing {len(pans_to_verify)} PAN verifications...")
results = []
for i, pan_data in enumerate(pans_to_verify, 1):
print(f"\n [{i}/{len(pans_to_verify)}] Verifying {pan_data['pan']}...")
try:
result = await client.call_tool(
tool_name="verify_pan",
params={
**pan_data,
"consent": "Y",
"reason": "Batch verification"
}
)
results.append({
"pan": pan_data["pan"],
"status": result.get("status"),
"name_match": result.get("name_match"),
"success": True
})
print(f" ✓ Status: {result.get('status')}")
except Exception as e:
results.append({
"pan": pan_data["pan"],
"error": str(e),
"success": False
})
print(f" ✗ Error: {e}")
# Summary
print("\n" + "-"*60)
print("Batch Processing Summary:")
successful = sum(1 for r in results if r.get("success"))
print(f" Total: {len(results)}")
print(f" Successful: {successful}")
print(f" Failed: {len(results) - successful}")
except Exception as e:
print(f"\n✗ Error: {e}")
finally:
await client.disconnect()
async def main():
"""Run all examples."""
print("\n" + "="*60)
print("MCP KYC Server - Client Examples")
print("="*60)
# Run examples
await example_list_tools()
await example_pan_verification()
await example_pan_aadhaar_link()
await example_error_handling()
await example_batch_operations()
print("\n" + "="*60)
print("All examples completed!")
print("="*60 + "\n")
if __name__ == "__main__":
# Run examples
asyncio.run(main())