#!/usr/bin/env python3
"""
Example client demonstrating CSV MCP server usage.
"""
import asyncio
import json
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client
async def main():
"""Demonstrate CSV MCP server functionality."""
# Server parameters for stdio connection
server_params = StdioServerParameters(
command="uv",
args=["run", "csv-mcp-server"]
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
print("š CSV MCP Server Demo")
print("=" * 50)
# List available tools
tools = await session.list_tools()
print(f"\nš Available tools ({len(tools.tools)}):")
for tool in tools.tools:
print(f" - {tool.name}: {tool.description}")
# Create a sample CSV file
print("\nš Creating sample CSV file...")
result = await session.call_tool(
"create_csv",
arguments={
"filename": "employees",
"headers": ["name", "age", "department", "salary"],
"data": [
["John Doe", 30, "Engineering", 75000],
["Jane Smith", 28, "Marketing", 65000],
["Bob Johnson", 35, "Engineering", 80000],
["Alice Brown", 32, "Sales", 60000],
["Charlie Wilson", 29, "Engineering", 70000]
]
}
)
print(f"Result: {json.dumps(result.content[0].text, indent=2)}")
# Read the CSV file
print("\nš Reading CSV file...")
result = await session.call_tool(
"read_csv",
arguments={"filename": "employees", "limit": 3}
)
print(f"First 3 rows: {json.dumps(result.content[0].text, indent=2)}")
# Get file information
print("\nš Getting file information...")
result = await session.call_tool(
"get_info",
arguments={"filename": "employees"}
)
print(f"File info: {json.dumps(result.content[0].text, indent=2)}")
# Get statistics
print("\nš Getting statistics...")
result = await session.call_tool(
"get_statistics",
arguments={"filename": "employees"}
)
print(f"Statistics: {json.dumps(result.content[0].text, indent=2)}")
# Filter data
print("\nš Filtering data (Engineering department)...")
result = await session.call_tool(
"filter_data",
arguments={
"filename": "employees",
"conditions": {"department": "Engineering"}
}
)
print(f"Filtered data: {json.dumps(result.content[0].text, indent=2)}")
# Sort data
print("\nš Sorting by salary (descending)...")
result = await session.call_tool(
"sort_data",
arguments={
"filename": "employees",
"columns": "salary",
"ascending": False
}
)
print(f"Sorted data: {json.dumps(result.content[0].text, indent=2)}")
# Group data
print("\nš Grouping by department...")
result = await session.call_tool(
"group_data",
arguments={
"filename": "employees",
"group_by": "department",
"aggregations": {"salary": "mean", "age": "mean"}
}
)
print(f"Grouped data: {json.dumps(result.content[0].text, indent=2)}")
# Add a new row
print("\nā Adding new employee...")
result = await session.call_tool(
"add_row",
arguments={
"filename": "employees",
"row_data": {
"name": "Eve Davis",
"age": 26,
"department": "HR",
"salary": 55000
}
}
)
print(f"Add result: {json.dumps(result.content[0].text, indent=2)}")
# Update a cell
print("\nāļø Updating salary for first employee...")
result = await session.call_tool(
"update_csv",
arguments={
"filename": "employees",
"row_index": 0,
"column": "salary",
"value": 78000
}
)
print(f"Update result: {json.dumps(result.content[0].text, indent=2)}")
# Validate data
print("\nā
Validating data integrity...")
result = await session.call_tool(
"validate_data",
arguments={"filename": "employees"}
)
print(f"Validation result: {json.dumps(result.content[0].text, indent=2)}")
# List all CSV files
print("\nš Listing all CSV files...")
result = await session.call_tool("list_csv_files", arguments={})
print(f"CSV files: {json.dumps(result.content[0].text, indent=2)}")
# Test resources
print("\nšļø Testing resources...")
resources = await session.list_resources()
print(f"Available resources: {len(resources.resources)}")
for resource in resources.resources[:3]: # Show first 3
print(f" - {resource.uri}")
# Test prompts
print("\nš Testing prompts...")
prompts = await session.list_prompts()
print(f"Available prompts: {len(prompts.prompts)}")
for prompt in prompts.prompts:
print(f" - {prompt.name}: {prompt.description}")
# Get analysis prompt
print("\nš Getting analysis prompt...")
prompt_result = await session.get_prompt(
"analyze_csv",
arguments={
"filename": "employees",
"analysis_type": "basic"
}
)
print("Analysis prompt:")
print(prompt_result.messages[0].content.text[:300] + "...")
print("\nāØ Demo completed successfully!")
if __name__ == "__main__":
asyncio.run(main())