#!/usr/bin/env python3
"""
Cost Explorer MCP Server wrapped for AgentCore Runtime deployment.
This file wraps the existing Cost Explorer MCP Server with BedrockAgentCoreApp
to enable deployment to AgentCore Runtime with MCP protocol support.
"""
import asyncio
import json
import os
import sys
from typing import Any, Dict
from bedrock_agentcore.runtime import BedrockAgentCoreApp
from awslabs.cost_explorer_mcp_server.comparison_handler import (
get_cost_and_usage_comparisons,
get_cost_comparison_drivers,
)
from awslabs.cost_explorer_mcp_server.cost_usage_handler import get_cost_and_usage
from awslabs.cost_explorer_mcp_server.forecasting_handler import get_cost_forecast
from awslabs.cost_explorer_mcp_server.metadata_handler import (
get_dimension_values,
get_tag_values,
)
from awslabs.cost_explorer_mcp_server.utility_handler import get_today_date
from loguru import logger
from mcp.server.fastmcp import FastMCP
# Configure Loguru logging
logger.remove()
logger.add(sys.stderr, level=os.getenv('FASTMCP_LOG_LEVEL', 'WARNING'))
# Define server instructions
SERVER_INSTRUCTIONS = """
# AWS Cost Explorer MCP Server
## IMPORTANT: Each API call costs $0.01 - use filters and specific date ranges to minimize charges.
## Critical Rules
- Comparison periods: exactly 1 month, start on day 1 (e.g., "2025-04-01" to "2025-05-01")
- UsageQuantity: Recommended to filter by USAGE_TYPE, USAGE_TYPE_GROUP or results are meaningless
- When user says "last X months": Use complete calendar months, not partial periods
- get_cost_comparison_drivers: returns only top 10 most significant drivers
## Query Pattern Mapping
| User Query Pattern | Recommended Tool | Notes |
|-------------------|-----------------|-------|
| "What were my costs for..." | get_cost_and_usage | Use for historical cost analysis |
| "How much did I spend on..." | get_cost_and_usage | Filter by service/region as needed |
| "Show me costs by..." | get_cost_and_usage | Set group_by parameter accordingly |
| "Compare costs between..." | get_cost_and_usage_comparisons | Ensure exactly 1 month periods |
| "Why did my costs change..." | get_cost_comparison_drivers | Returns top 10 drivers only |
| "What caused my bill to..." | get_cost_comparison_drivers | Good for root cause analysis |
| "Predict/forecast my costs..." | get_cost_forecast | Works best with specific services |
| "What will I spend on..." | get_cost_forecast | Can filter by dimension |
## Cost Optimization Tips
- Always use specific date ranges rather than broad periods
- Filter by specific services when possible to reduce data processed
- For usage metrics, always filter by USAGE_TYPE or USAGE_TYPE_GROUP to get meaningful results
- Combine related questions into a single query where possible
"""
# Create FastMCP server with streamable HTTP support for Agentcore Runtime
mcp_app = FastMCP(
name='Cost Explorer MCP Server',
instructions=SERVER_INSTRUCTIONS,
host="0.0.0.0",
port=8000,
stateless_http=True
)
# Register all tools with the MCP app
mcp_app.tool('get_today_date')(get_today_date)
mcp_app.tool('get_dimension_values')(get_dimension_values)
mcp_app.tool('get_tag_values')(get_tag_values)
mcp_app.tool('get_cost_forecast')(get_cost_forecast)
mcp_app.tool('get_cost_and_usage_comparisons')(get_cost_and_usage_comparisons)
mcp_app.tool('get_cost_comparison_drivers')(get_cost_comparison_drivers)
mcp_app.tool('get_cost_and_usage')(get_cost_and_usage)
# Create AgentCore Runtime app
app = BedrockAgentCoreApp()
@app.entrypoint
def invoke(payload: Dict[str, Any], context: Any) -> Dict[str, Any]:
"""
AgentCore Runtime entrypoint for the Cost Explorer MCP Server.
This function handles incoming requests from AgentCore Runtime and
delegates them to the MCP server running in streamable HTTP mode.
Args:
payload: The request payload from AgentCore Runtime
context: AgentCore Runtime context
Returns:
Dictionary containing the response from the MCP server
"""
try:
logger.info(f"AgentCore Runtime invoke called with payload: {payload}")
# Extract the prompt or message from the payload
user_message = payload.get("prompt", "")
if not user_message:
user_message = payload.get("message", "Hello! I'm the Cost Explorer MCP Server.")
# For AgentCore Runtime with MCP protocol, we return information about the MCP server
response = {
"result": f"Cost Explorer MCP Server is running and ready to serve MCP requests.",
"message": user_message,
"mcp_endpoint": "http://0.0.0.0:8000/mcp",
"available_tools": [
"get_today_date",
"get_dimension_values",
"get_tag_values",
"get_cost_forecast",
"get_cost_and_usage_comparisons",
"get_cost_comparison_drivers",
"get_cost_and_usage"
],
"instructions": SERVER_INSTRUCTIONS,
"status": "ready"
}
logger.info(f"AgentCore Runtime response: {response}")
return response
except Exception as e:
logger.error(f"Error in AgentCore Runtime invoke: {e}")
return {
"error": str(e),
"status": "error"
}
def start_mcp_server():
"""Start the MCP server in the background."""
try:
logger.info("Starting Cost Explorer MCP Server on port 8000")
# Start the MCP server with streamable HTTP transport
mcp_app.run(transport="streamable-http")
except Exception as e:
logger.error(f"Failed to start MCP server: {e}")
raise
if __name__ == "__main__":
# When running locally, start both the MCP server and AgentCore app
import threading
# Start MCP server in a separate thread
mcp_thread = threading.Thread(target=start_mcp_server, daemon=True)
mcp_thread.start()
logger.info("Starting AgentCore Runtime app")
# Start the AgentCore Runtime app
app.run()
