#!/usr/bin/env python3
"""
Natural Language Demo for Cost Explorer MCP Server.
This script demonstrates how the MCP server would handle real-world business questions
with simulated responses showing the expected behavior with proper AWS permissions.
"""
import asyncio
import json
import os
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
class CostExplorerDemo:
"""Demo class showing natural language scenarios with expected responses."""
def __init__(self, mcp_url, headers):
self.mcp_url = mcp_url
self.headers = headers
self.session = None
async def connect(self):
"""Connect to the MCP server."""
self.client_context = streamablehttp_client(
self.mcp_url,
self.headers,
timeout=60,
terminate_on_close=False
)
self.streams = await self.client_context.__aenter__()
read_stream, write_stream, _ = self.streams
self.session = ClientSession(read_stream, write_stream)
await self.session.__aenter__()
await self.session.initialize()
print("ā
Connected to Cost Explorer MCP Server")
async def disconnect(self):
"""Disconnect from the MCP server."""
if self.session:
await self.session.__aexit__(None, None, None)
if hasattr(self, 'client_context'):
await self.client_context.__aexit__(None, None, None)
async def demo_tool_call(self, tool_name, arguments, description, expected_response):
"""Demo a tool call with expected response."""
print(f"š§ Tool: {tool_name}")
print(f" š Business Context: {description}")
print(f" š Parameters: {json.dumps(arguments, indent=6)}")
try:
result = await self.session.call_tool(tool_name, arguments)
if result.content and len(result.content) > 0:
content = result.content[0].text
try:
parsed = json.loads(content)
if "error" in parsed:
print(f" ā ļø Current Status: AWS permissions needed")
print(f" šÆ Expected Response: {expected_response}")
else:
print(f" ā
Actual Response: {json.dumps(parsed, indent=6)}")
except:
print(f" ā
Response: {content}")
else:
print(f" šÆ Expected Response: {expected_response}")
except Exception as e:
print(f" ā ļø Current Status: {str(e)[:100]}...")
print(f" šÆ Expected Response: {expected_response}")
def print_scenario(self, scenario_num, title, question):
"""Print a formatted scenario header."""
print(f"\n{'='*90}")
print(f"š SCENARIO {scenario_num}: {title}")
print(f"{'='*90}")
print(f"ā Business Question: \"{question}\"")
print(f"š¤ MCP Server Analysis:")
async def run_natural_language_demo():
"""Run natural language demo scenarios."""
# Setup connection
bearer_token = os.getenv('BEARER_TOKEN')
agent_arn = None
if os.path.exists('agent_arn.txt'):
with open('agent_arn.txt', 'r') as f:
agent_arn = f.read().strip()
if not bearer_token or not agent_arn:
print("ā Missing BEARER_TOKEN or agent ARN")
return
# Construct MCP URL
encoded_arn = agent_arn.replace(':', '%3A').replace('/', '%2F')
mcp_url = f"https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/{encoded_arn}/invocations?qualifier=DEFAULT"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {bearer_token}",
"Accept": "application/json"
}
print("šÆ Cost Explorer MCP Server - Natural Language Business Scenarios")
print("=" * 90)
print("This demonstrates how natural language questions map to MCP tool calls")
print("and shows expected responses with proper AWS Cost Explorer permissions.")
demo = CostExplorerDemo(mcp_url, headers)
try:
await demo.connect()
# SCENARIO 1: Context Setting
demo.print_scenario(1, "Context & Date Analysis",
"What's today's date for our cost analysis?")
await demo.demo_tool_call(
"get_today_date",
{},
"Establishing temporal context for relative date queries",
"Current date and month for cost analysis context"
)
# SCENARIO 2: Service Discovery
demo.print_scenario(2, "AWS Service Discovery",
"What AWS services are we using that generate costs?")
await demo.demo_tool_call(
"get_dimension_values",
{
"date_range": {
"start_date": "2025-12-01",
"end_date": "2025-12-31"
},
"dimension": {"Key": "SERVICE"}
},
"Discovering all AWS services with cost data in the account",
"List of AWS services: EC2, S3, RDS, Lambda, CloudWatch, etc."
)
# SCENARIO 3: Monthly Cost Breakdown
demo.print_scenario(3, "Monthly Service Costs",
"Show me our December 2025 AWS costs broken down by service")
await demo.demo_tool_call(
"get_cost_and_usage",
{
"date_range": {
"start_date": "2025-12-01",
"end_date": "2025-12-31"
},
"granularity": "MONTHLY",
"group_by": {"Type": "DIMENSION", "Key": "SERVICE"},
"metric": "UnblendedCost"
},
"Breaking down total monthly costs by AWS service",
"Service costs: EC2: $1,234, S3: $567, RDS: $890, etc."
)
# SCENARIO 4: Regional Analysis
demo.print_scenario(4, "Multi-Region Cost Analysis",
"Which AWS regions are we spending money in and how much?")
await demo.demo_tool_call(
"get_cost_and_usage",
{
"date_range": {
"start_date": "2025-12-01",
"end_date": "2025-12-31"
},
"granularity": "MONTHLY",
"group_by": {"Type": "DIMENSION", "Key": "REGION"},
"metric": "UnblendedCost"
},
"Analyzing cost distribution across AWS regions",
"Regional costs: us-east-1: $2,345, us-west-2: $1,234, eu-west-1: $567"
)
# SCENARIO 5: EC2 Deep Dive
demo.print_scenario(5, "EC2 Instance Analysis",
"What types of EC2 instances are we running and what do they cost?")
await demo.demo_tool_call(
"get_cost_and_usage",
{
"date_range": {
"start_date": "2025-12-01",
"end_date": "2025-12-31"
},
"granularity": "MONTHLY",
"group_by": {"Type": "DIMENSION", "Key": "USAGE_TYPE"},
"filter_expression": {
"Dimensions": {
"Key": "SERVICE",
"Values": ["Amazon Elastic Compute Cloud - Compute"],
"MatchOptions": ["EQUALS"]
}
},
"metric": "UnblendedCost"
},
"Analyzing EC2 costs by instance type and usage pattern",
"EC2 usage: t3.medium: $456, m5.large: $789, c5.xlarge: $234"
)
# SCENARIO 6: Cost Comparison
demo.print_scenario(6, "Month-over-Month Comparison",
"How do our November vs December 2025 costs compare?")
await demo.demo_tool_call(
"get_cost_and_usage_comparisons",
{
"baseline_date_range": {
"start_date": "2025-11-01",
"end_date": "2025-12-01"
},
"comparison_date_range": {
"start_date": "2025-12-01",
"end_date": "2026-01-01"
},
"metric_for_comparison": "UnblendedCost",
"group_by": {"Type": "DIMENSION", "Key": "SERVICE"}
},
"Comparing costs between November and December 2025",
"Cost changes: Total +15%, EC2 +25%, S3 -10%, RDS +5%"
)
# SCENARIO 7: Cost Driver Analysis
demo.print_scenario(7, "Cost Change Investigation",
"What caused our AWS bill to increase from November to December?")
await demo.demo_tool_call(
"get_cost_comparison_drivers",
{
"baseline_date_range": {
"start_date": "2025-11-01",
"end_date": "2025-12-01"
},
"comparison_date_range": {
"start_date": "2025-12-01",
"end_date": "2026-01-01"
},
"metric_for_comparison": "UnblendedCost",
"group_by": {"Type": "DIMENSION", "Key": "SERVICE"}
},
"Identifying top 10 cost change drivers between months",
"Top drivers: 1) EC2 m5.large +$500, 2) RDS storage +$200, 3) S3 requests +$100"
)
# SCENARIO 8: Storage Optimization
demo.print_scenario(8, "S3 Storage Cost Analysis",
"How much are we spending on S3 and what storage classes?")
await demo.demo_tool_call(
"get_cost_and_usage",
{
"date_range": {
"start_date": "2025-12-01",
"end_date": "2025-12-31"
},
"granularity": "MONTHLY",
"group_by": {"Type": "DIMENSION", "Key": "USAGE_TYPE"},
"filter_expression": {
"Dimensions": {
"Key": "SERVICE",
"Values": ["Amazon Simple Storage Service"],
"MatchOptions": ["EQUALS"]
}
},
"metric": "UnblendedCost"
},
"Breaking down S3 costs by storage class and operations",
"S3 costs: Standard: $234, IA: $123, Glacier: $45, Requests: $67"
)
# SCENARIO 9: Tag-based Analysis
demo.print_scenario(9, "Project Cost Allocation",
"What Environment tags do we use for cost tracking?")
await demo.demo_tool_call(
"get_tag_values",
{
"date_range": {
"start_date": "2025-12-01",
"end_date": "2025-12-31"
},
"tag_key": "Environment"
},
"Discovering Environment tag values for project cost allocation",
"Environment tags: Production, Staging, Development, Testing"
)
# SCENARIO 10: Budget Forecasting
demo.print_scenario(10, "3-Month Cost Forecast",
"What will our AWS costs look like for Q1 2026?")
await demo.demo_tool_call(
"get_cost_forecast",
{
"date_range": {
"start_date": "2026-01-01",
"end_date": "2026-04-01"
},
"granularity": "MONTHLY",
"metric": "UNBLENDED_COST",
"prediction_interval_level": 80
},
"Generating quarterly cost forecast with confidence intervals",
"Q1 2026 forecast: Jan: $3,200Ā±$400, Feb: $3,100Ā±$350, Mar: $3,300Ā±$450"
)
# SCENARIO 11: Usage Metrics
demo.print_scenario(11, "EC2 Usage Hours",
"How many compute hours did we use in December?")
await demo.demo_tool_call(
"get_cost_and_usage",
{
"date_range": {
"start_date": "2025-12-01",
"end_date": "2025-12-31"
},
"granularity": "MONTHLY",
"group_by": {"Type": "DIMENSION", "Key": "USAGE_TYPE"},
"filter_expression": {
"And": [
{
"Dimensions": {
"Key": "SERVICE",
"Values": ["Amazon Elastic Compute Cloud - Compute"],
"MatchOptions": ["EQUALS"]
}
},
{
"Dimensions": {
"Key": "USAGE_TYPE_GROUP",
"Values": ["EC2: Running Hours"],
"MatchOptions": ["EQUALS"]
}
}
]
},
"metric": "UsageQuantity"
},
"Measuring actual compute hours by instance type",
"Usage hours: t3.medium: 2,160 hrs, m5.large: 1,440 hrs, c5.xlarge: 720 hrs"
)
# SCENARIO 12: Service Ranking
demo.print_scenario(12, "Top Cost Services",
"Rank our AWS services by cost over the last quarter")
await demo.demo_tool_call(
"get_cost_and_usage",
{
"date_range": {
"start_date": "2025-10-01",
"end_date": "2025-12-31"
},
"granularity": "MONTHLY",
"group_by": {"Type": "DIMENSION", "Key": "SERVICE"},
"metric": "UnblendedCost"
},
"Ranking AWS services by total quarterly spend",
"Service ranking: 1) EC2: $4,567, 2) RDS: $2,345, 3) S3: $1,234, 4) Lambda: $567"
)
print(f"\n{'='*90}")
print("š NATURAL LANGUAGE DEMO COMPLETE!")
print("=" * 90)
print("ā
Demonstrated 12 real-world business scenarios")
print("ā
Showed natural language ā MCP tool mapping")
print("ā
Covered all 7 Cost Explorer tools with proper parameters")
print("ā
Included expected responses for business context")
print("\nš” Key Business Value:")
print(" šÆ Complex questions ā Simple tool calls")
print(" š Natural language ā Structured data queries")
print(" š Business context ā Technical parameters")
print(" š Cost insights ā Actionable recommendations")
print("\nš§ Technical Insights:")
print(" ā¢ Proper date range handling for business periods")
print(" ā¢ Intelligent filtering based on question context")
print(" ā¢ Appropriate grouping for meaningful analysis")
print(" ā¢ Error handling with business-friendly messages")
print("\nš Next Steps:")
print(" 1. Add AWS Cost Explorer IAM permissions")
print(" 2. Configure cost allocation tags")
print(" 3. Set up automated cost monitoring")
print(" 4. Integrate with business intelligence tools")
except Exception as e:
print(f"ā Error during demo: {e}")
finally:
await demo.disconnect()
if __name__ == "__main__":
print("š Cost Explorer MCP Server - Natural Language Business Demo")
print("This demonstrates real-world business scenarios and expected responses.\n")
if not os.getenv('BEARER_TOKEN'):
print("ā Please set BEARER_TOKEN environment variable")
exit(1)
asyncio.run(run_natural_language_demo())
