Trade Surveillance Support MCP Server

Instructions

Parse a user inquiry email to extract key information for investigation.

This tool analyzes email content and extracts:
- Inquiry type (trade issue, report request, data verification, etc.)
- Related trade IDs or account numbers
- Time period of interest
- Priority level
- Required actions

Args:
    email_content: The full text content of the user's inquiry email
    
Returns:
    A dictionary containing parsed information including inquiry_type,
    trade_ids, time_period, priority, and suggested_actions

Implementation Reference

• trade_surveillance_mcp/server.py:32-190 (handler)

  The async function parse_email_inquiry parses the email content to extract trade IDs, account numbers, time periods, inquiry type, priority level, and generates suggested actions using regex patterns and keyword matching.

  async def parse_email_inquiry(email_content: str) -> dict[str, Any]:
      """
      Parse a user inquiry email to extract key information for investigation.
      
      This tool analyzes email content and extracts:
      - Inquiry type (trade issue, report request, data verification, etc.)
      - Related trade IDs or account numbers
      - Time period of interest
      - Priority level
      - Required actions
      
      Args:
          email_content: The full text content of the user's inquiry email
          
      Returns:
          A dictionary containing parsed information including inquiry_type,
          trade_ids, time_period, priority, and suggested_actions
      """
      import re
      from datetime import datetime, timedelta
      
      content_lower = email_content.lower()
      
      # Extract trade IDs (common patterns: TRD123456, TRADE-123456, T-123456, #123456)
      trade_id_patterns = [
          r'\bTRD[\-_]?\d{5,10}\b',
          r'\bTRADE[\-_]?\d{5,10}\b',
          r'\bT[\-_]\d{5,10}\b',
          r'#\d{5,10}\b',
          r'\btrade\s+(?:id|number|ref)[\s:]+(\d{5,10})\b'
      ]
      trade_ids = []
      for pattern in trade_id_patterns:
          matches = re.findall(pattern, email_content, re.IGNORECASE)
          trade_ids.extend(matches)
      trade_ids = list(set(trade_ids))  # Remove duplicates
      
      # Extract account numbers (common patterns: ACC123456, ACCT-123456, Account: 123456)
      account_patterns = [
          r'\bACC(?:T)?[\-_]?\d{5,10}\b',
          r'\baccount[\s:]+(\d{5,10})\b'
      ]
      account_numbers = []
      for pattern in account_patterns:
          matches = re.findall(pattern, email_content, re.IGNORECASE)
          account_numbers.extend(matches)
      account_numbers = list(set(account_numbers))
      
      # Extract time periods (dates, ranges, relative times)
      time_period = None
      date_patterns = [
          r'\b(\d{1,2}[-/]\d{1,2}[-/]\d{2,4})\b',  # MM/DD/YYYY or DD-MM-YYYY
          r'\b(\d{4}[-/]\d{1,2}[-/]\d{1,2})\b',    # YYYY-MM-DD
          r'\b(jan|feb|mar|apr|may|jun|jul|aug|sep|oct|nov|dec)[a-z]*\s+\d{1,2},?\s+\d{4}\b',
          r'\b\d{1,2}\s+(jan|feb|mar|apr|may|jun|jul|aug|sep|oct|nov|dec)[a-z]*\s+\d{4}\b'
      ]
      
      dates_found = []
      for pattern in date_patterns:
          dates_found.extend(re.findall(pattern, email_content, re.IGNORECASE))
      
      # Check for relative time references
      if 'last week' in content_lower or 'past week' in content_lower:
          time_period = "last_week"
      elif 'last month' in content_lower or 'past month' in content_lower:
          time_period = "last_month"
      elif 'yesterday' in content_lower:
          time_period = "yesterday"
      elif 'today' in content_lower or 'this morning' in content_lower:
          time_period = "today"
      elif 'last 7 days' in content_lower or 'past 7 days' in content_lower:
          time_period = "last_7_days"
      elif 'last 30 days' in content_lower or 'past 30 days' in content_lower:
          time_period = "last_30_days"
      elif dates_found:
          time_period = dates_found[0] if isinstance(dates_found[0], str) else str(dates_found[0])
      
      # Determine inquiry type based on keywords
      inquiry_type = "general_inquiry"
      inquiry_keywords = {
          "trade_issue": ["trade error", "failed trade", "trade problem", "transaction failed", "execution issue"],
          "report_request": ["generate report", "need report", "send report", "report for", "can you provide"],
          "data_verification": ["verify", "check data", "confirm", "validate", "reconcile"],
          "settlement_inquiry": ["settlement", "settle", "payment", "clearing"],
          "compliance_check": ["compliance", "audit", "regulatory", "regulation"],
          "position_inquiry": ["position", "holdings", "balance", "exposure"],
          "transaction_history": ["transaction history", "trade history", "past trades", "historical"]
      }
      
      for inq_type, keywords in inquiry_keywords.items():
          if any(keyword in content_lower for keyword in keywords):
              inquiry_type = inq_type
              break
      
      # Determine priority based on urgency indicators
      priority = "medium"
      high_priority_words = ["urgent", "asap", "immediately", "critical", "emergency", "high priority"]
      low_priority_words = ["when you can", "no rush", "low priority", "whenever"]
      
      if any(word in content_lower for word in high_priority_words):
          priority = "high"
      elif any(word in content_lower for word in low_priority_words):
          priority = "low"
      
      # Generate suggested actions based on inquiry type
      suggested_actions = []
      if inquiry_type == "report_request":
          suggested_actions = [
              "Search for relevant report config files",
              "Identify appropriate Java report generator",
              "Execute report generation",
              "Send report to requester"
          ]
      elif inquiry_type == "trade_issue":
          suggested_actions = [
              "Search trade transaction records",
              "Check for error logs",
              "Verify trade details",
              "Generate diagnostic report"
          ]
      elif inquiry_type == "data_verification":
          suggested_actions = [
              "Query relevant data sources",
              "Run reconciliation checks",
              "Generate verification report"
          ]
      elif inquiry_type == "settlement_inquiry":
          suggested_actions = [
              "Search settlement config files",
              "Check settlement status",
              "Generate settlement report"
          ]
      elif inquiry_type == "compliance_check":
          suggested_actions = [
              "Search compliance config files",
              "Run audit checks",
              "Generate compliance report"
          ]
      else:
          suggested_actions = [
              "Analyze inquiry details",
              "Search for relevant config files",
              "Identify appropriate action"
          ]
      
      result = {
          "status": "parsed",
          "inquiry_type": inquiry_type,
          "trade_ids": trade_ids,
          "account_numbers": account_numbers,
          "time_period": time_period,
          "priority": priority,
          "suggested_actions": suggested_actions,
          "raw_content_preview": email_content[:300] + "..." if len(email_content) > 300 else email_content,
          "content_length": len(email_content)
      }
      
      logger.info(f"Parsed email inquiry: {inquiry_type}, Priority: {priority}, Found {len(trade_ids)} trade IDs")
      return result

• trade_surveillance_mcp/server.py:31-31 (registration)

  The @mcp.tool() decorator registers the parse_email_inquiry function as an MCP tool on the FastMCP server instance.

  @mcp.tool()