#!/usr/bin/env python3
"""
Example showing how to use the CSV MCP server for data analysis workflows.
"""
import asyncio
import json
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client
async def sales_analysis_example():
"""Demonstrate a sales data analysis workflow."""
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("š Sales Data Analysis Example")
print("=" * 40)
# Create sales data
print("\n1. Creating sales data...")
await session.call_tool(
"create_csv",
arguments={
"filename": "sales_2024",
"headers": ["date", "product", "category", "quantity", "unit_price", "customer_type"],
"data": [
["2024-01-15", "Laptop Pro", "Electronics", 2, 1299.99, "Business"],
["2024-01-16", "Wireless Mouse", "Electronics", 5, 49.99, "Consumer"],
["2024-01-17", "Office Chair", "Furniture", 1, 299.99, "Business"],
["2024-01-18", "Desk Lamp", "Furniture", 3, 79.99, "Consumer"],
["2024-01-19", "Laptop Pro", "Electronics", 1, 1299.99, "Consumer"],
["2024-01-20", "Standing Desk", "Furniture", 2, 599.99, "Business"],
["2024-01-21", "Keyboard", "Electronics", 4, 89.99, "Consumer"],
["2024-01-22", "Monitor", "Electronics", 3, 249.99, "Business"],
["2024-01-23", "Office Chair", "Furniture", 2, 299.99, "Consumer"],
["2024-01-24", "Wireless Mouse", "Electronics", 8, 49.99, "Business"]
]
}
)
# Calculate total revenue per row (quantity * unit_price)
print("\n2. Adding calculated revenue column...")
# First, let's read the data to calculate revenues
result = await session.call_tool("read_csv", arguments={"filename": "sales_2024"})
data = json.loads(result.content[0].text)["data"]
for i, row in enumerate(data):
revenue = row["quantity"] * row["unit_price"]
await session.call_tool(
"add_row",
arguments={
"filename": "sales_with_revenue",
"row_data": {
**row,
"revenue": revenue
}
}
) if i == 0 else None
if i == 0:
# Create the new file with revenue column
await session.call_tool(
"create_csv",
arguments={
"filename": "sales_with_revenue",
"headers": ["date", "product", "category", "quantity", "unit_price", "customer_type", "revenue"]
}
)
await session.call_tool(
"add_row",
arguments={
"filename": "sales_with_revenue",
"row_data": {
**row,
"revenue": revenue
}
}
)
# Analyze by category
print("\n3. Analyzing sales by category...")
result = await session.call_tool(
"group_data",
arguments={
"filename": "sales_with_revenue",
"group_by": "category",
"aggregations": {
"revenue": "sum",
"quantity": "sum"
}
}
)
print("Revenue by Category:")
print(json.dumps(json.loads(result.content[0].text)["grouped_data"], indent=2))
# Find top products by revenue
print("\n4. Finding top products by revenue...")
result = await session.call_tool(
"sort_data",
arguments={
"filename": "sales_with_revenue",
"columns": "revenue",
"ascending": False,
"limit": 5
}
)
top_products = json.loads(result.content[0].text)["sorted_data"]
print("Top 5 Sales by Revenue:")
for product in top_products:
print(f" {product['product']}: ${product['revenue']:.2f}")
# Analyze customer types
print("\n5. Analyzing by customer type...")
result = await session.call_tool(
"group_data",
arguments={
"filename": "sales_with_revenue",
"group_by": "customer_type",
"aggregations": {
"revenue": "sum",
"quantity": "count"
}
}
)
customer_analysis = json.loads(result.content[0].text)["grouped_data"]
print("Sales by Customer Type:")
for customer in customer_analysis:
print(f" {customer['customer_type']}: ${customer['revenue']:.2f} ({customer['quantity']} transactions)")
# Filter high-value transactions
print("\n6. Finding high-value transactions (>$500)...")
result = await session.call_tool(
"filter_data",
arguments={
"filename": "sales_with_revenue",
"conditions": {
"revenue": {"gt": 500}
}
}
)
high_value = json.loads(result.content[0].text)
print(f"Found {high_value['filtered_rows']} high-value transactions")
for transaction in high_value["filtered_data"]:
print(f" {transaction['product']}: ${transaction['revenue']:.2f}")
# Get statistics
print("\n7. Getting statistical summary...")
result = await session.call_tool(
"get_statistics",
arguments={"filename": "sales_with_revenue"}
)
stats = json.loads(result.content[0].text)["statistics"]
print("Revenue Statistics:")
if "revenue" in stats:
print(f" Mean: ${stats['revenue']['mean']:.2f}")
print(f" Median: ${stats['revenue']['50%']:.2f}")
print(f" Max: ${stats['revenue']['max']:.2f}")
print(f" Min: ${stats['revenue']['min']:.2f}")
# Generate analysis prompt
print("\n8. Generating analysis recommendations...")
result = await session.get_prompt(
"analyze_csv",
arguments={
"filename": "sales_with_revenue",
"analysis_type": "statistical"
}
)
print("\nAI Analysis Prompt Generated:")
print(result.messages[0].content.text[:500] + "...")
print("\nāØ Sales analysis workflow completed!")
if __name__ == "__main__":
asyncio.run(sales_analysis_example())