AWS Cost Explorer MCP Server

Instructions

Retrieve EC2 spend for the last day using standard AWS Cost Explorer API.

Returns:
    Dict[str, Any]: The raw response from the AWS Cost Explorer API, or None if an error occurs.

Implementation Reference

• server.py:644-733 (handler)

  The main handler function for the get_ec2_spend_last_day MCP tool. Decorated with @mcp.tool() which registers it with the FastMCP server. Implements logic to query AWS Cost Explorer for EC2 compute costs grouped by instance type for the last day.

  @mcp.tool()
  async def get_ec2_spend_last_day(params: EC2Params) -> Dict[str, Any]:
      """
      Retrieve EC2 spend for the last day using standard AWS Cost Explorer API.
      
      Returns:
          Dict[str, Any]: The raw response from the AWS Cost Explorer API, or None if an error occurs.
      """
      print(f"get_ec2_spend_last_day, params={params}")
      # Initialize the Cost Explorer client
      ce_client = get_aws_service_boto3_client("ce", params.aws_account_id, params.region)
  
      
      # Calculate the time period - last day
      end_date = datetime.now().strftime('%Y-%m-%d')
      start_date = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
      
      try:
          # Make the API call using get_cost_and_usage (standard API)
          response = ce_client.get_cost_and_usage(
              TimePeriod={
                  'Start': start_date,
                  'End': end_date
              },
              Granularity='DAILY',
              Filter={
                  'Dimensions': {
                      'Key': 'SERVICE',
                      'Values': [
                          'Amazon Elastic Compute Cloud - Compute'
                      ]
                  }
              },
              Metrics=[
                  'UnblendedCost',
                  'UsageQuantity'
              ],
              GroupBy=[
                  {
                      'Type': 'DIMENSION',
                      'Key': 'INSTANCE_TYPE'
                  }
              ]
          )
          
          # Process and print the results
          print(f"EC2 Spend from {start_date} to {end_date}:")
          print("-" * 50)
          
          total_cost = 0.0
          
          if 'ResultsByTime' in response and response['ResultsByTime']:
              time_period_data = response['ResultsByTime'][0]
              
              if 'Groups' in time_period_data:
                  for group in time_period_data['Groups']:
                      instance_type = group['Keys'][0]
                      cost = float(group['Metrics']['UnblendedCost']['Amount'])
                      currency = group['Metrics']['UnblendedCost']['Unit']
                      usage = float(group['Metrics']['UsageQuantity']['Amount'])
                      
                      print(f"Instance Type: {instance_type}")
                      print(f"Cost: {cost:.4f} {currency}")
                      print(f"Usage: {usage:.2f}")
                      print("-" * 30)
                      
                      total_cost += cost
              
              # If no instance-level breakdown, show total
              if not time_period_data.get('Groups'):
                  if 'Total' in time_period_data:
                      total = time_period_data['Total']
                      cost = float(total['UnblendedCost']['Amount'])
                      currency = total['UnblendedCost']['Unit']
                      print(f"Total EC2 Cost: {cost:.4f} {currency}")
                  else:
                      print("No EC2 costs found for this period")
              else:
                  print(f"Total EC2 Cost: {total_cost:.4f} {currency if 'currency' in locals() else 'USD'}")
                  
              # Check if results are estimated
              if 'Estimated' in time_period_data:
                  print(f"Note: These results are {'estimated' if time_period_data['Estimated'] else 'final'}")
          
          return response
          
      except Exception as e:
          print(f"Error retrieving EC2 cost data: {str(e)}")
          return None

• server.py:52-68 (schema)

  Pydantic BaseModel schema used as input parameter type for the EC2-related tools, including get_ec2_spend_last_day. Defines validation for days, region, and optional cross-account ID.

  class EC2Params(BaseModel):
      """Parameters for retrieving EC2 Cost Explorer information."""
      days: int = Field(
          default=1,
          description="Number of days to look back for Bedrock logs",
          ge=1,
          le=90
      )
      region: str = Field(
          default="us-east-1",
          description="AWS region to retrieve logs from"
      )
      aws_account_id: Optional[str] = Field(        
          description="AWS account id (if different from the current AWS account) of the account for which to get the cost data",
          default=None
      )

• server.py:75-135 (helper)

  Utility function called by the handler to obtain a boto3 Cost Explorer ('ce') client, supporting cross-account access via role assumption.

  def get_aws_service_boto3_client(service: str, aws_account_id: Optional[str], region_name: str, account_b_role_name: Optional[str] = CROSS_ACCOUNT_ROLE_NAME):
      """
      Creates a boto3 client for the specified service in this current AWS account or in a different account
      if an account id is specified.
      
      Args:
          service (str): AWS service name (e.g., 'logs', 'cloudwatch')
          region_name (str): AWS region (e.g. 'us-east-1')
          aws_account_id (str): AWS account ID to access, this is the account in which the role is to be assumed
          account_b_role_name (str): IAM role name to assume
          
      Returns:
          boto3.client: Service client with assumed role credentials
      """
      try:
          this_account = boto3.client('sts').get_caller_identity()['Account']
          if aws_account_id is not None and this_account != aws_account_id:
              # the request is for a different account, we need to assume a role in that account
              print(f"Request is for a different account: {aws_account_id}, current account: {this_account}")
              # Create STS client
              sts_client = boto3.client('sts')
              current_identity = sts_client.get_caller_identity()
              print(f"Current identity: {current_identity}")
              
              # Define the role ARN
              role_arn = f"arn:aws:iam::{aws_account_id}:role/{account_b_role_name}"
              print(f"Attempting to assume role: {role_arn}")
              
              # Assume the role
              assumed_role = sts_client.assume_role(
                  RoleArn=role_arn,
                  RoleSessionName="CrossAccountSession"
              )
              
              # Extract temporary credentials
              credentials = assumed_role['Credentials']
              
              # Create client with assumed role credentials
              client = boto3.client(
                  service,
                  region_name=region_name,
                  aws_access_key_id=credentials['AccessKeyId'],
                  aws_secret_access_key=credentials['SecretAccessKey'],
                  aws_session_token=credentials['SessionToken']
              )
              
              print(f"Successfully created cross-account client for {service} in account {aws_account_id}")
              return client
          else:
              client = boto3.client(
                  service,
                  region_name=region_name
              )
              
              print(f"Successfully created client for {service} in the current AWS account {this_account}")
              return client
          
      except Exception as e:
          print(f"Error creating cross-account client for {service}: {e}")
          raise e