Polygon-io MCP Server

Instructions

Implementation Reference

• src/mcp_massive/server.py:1675-1693 (handler)

  The handler function decorated with @poly_mcp.tool that implements the tool logic by calling the Massive API client and formatting the JSON response to CSV.

  async def get_futures_contract_details(
      ticker: str,
      as_of: Optional[Union[str, date]] = None,
      params: Optional[Dict[str, Any]] = None,
  ) -> str:
      """
      Get details for a single futures contract at a specified point in time.
      """
      try:
          results = massive_client.get_futures_contract_details(
              ticker=ticker,
              as_of=as_of,
              params=params,
              raw=True,
          )
  
          return json_to_csv(results.data.decode("utf-8"))
      except Exception as e:
          return f"Error: {e}"

• src/mcp_massive/server.py:1675-1675 (registration)

  The @poly_mcp.tool decorator registers this function as an MCP tool.

  async def get_futures_contract_details(

• src/mcp_massive/server.py:1676-1678 (schema)

  The function signature defines the input schema (parameters) and output type for the tool.

  ticker: str,
  as_of: Optional[Union[str, date]] = None,
  params: Optional[Dict[str, Any]] = None,

• src/mcp_massive/formatters.py:7-104 (helper)

  The json_to_csv helper function used to format the API response from JSON to CSV for all tools including this one.

  def json_to_csv(json_input: str | dict) -> str:
      """
      Convert JSON to flattened CSV format.
  
      Args:
          json_input: JSON string or dict. If the JSON has a 'results' key containing
                     a list, it will be extracted. Otherwise, the entire structure
                     will be wrapped in a list for processing.
  
      Returns:
          CSV string with headers and flattened rows
      """
      # Parse JSON if it's a string
      if isinstance(json_input, str):
          try:
              data = json.loads(json_input)
          except json.JSONDecodeError:
              # If JSON parsing fails, return empty CSV
              return ""
      else:
          data = json_input
  
      if isinstance(data, dict) and "results" in data:
          results_value = data["results"]
          # Handle both list and single object responses
          if isinstance(results_value, list):
              records = results_value
          elif isinstance(results_value, dict):
              # Single object response (e.g., get_last_trade returns results as object)
              records = [results_value]
          else:
              records = [results_value]
      elif isinstance(data, dict) and "last" in data:
          # Handle responses with "last" key (e.g., get_last_trade, get_last_quote)
          records = [data["last"]] if isinstance(data["last"], dict) else [data]
      elif isinstance(data, list):
          records = data
      else:
          records = [data]
  
      # Only flatten dict records, skip non-dict items
      flattened_records = []
      for record in records:
          if isinstance(record, dict):
              flattened_records.append(_flatten_dict(record))
          else:
              # If it's not a dict, wrap it in a dict with a 'value' key
              flattened_records.append({"value": str(record)})
  
      if not flattened_records:
          return ""
  
      # Get all unique keys across all records (for consistent column ordering)
      all_keys = []
      seen = set()
      for record in flattened_records:
          if isinstance(record, dict):
              for key in record.keys():
                  if key not in seen:
                      all_keys.append(key)
                      seen.add(key)
  
      output = io.StringIO()
      writer = csv.DictWriter(output, fieldnames=all_keys, lineterminator="\n")
      writer.writeheader()
      writer.writerows(flattened_records)
  
      return output.getvalue()
  
  
  def _flatten_dict(
      d: dict[str, Any], parent_key: str = "", sep: str = "_"
  ) -> dict[str, Any]:
      """
      Flatten a nested dictionary by joining keys with separator.
  
      Args:
          d: Dictionary to flatten
          parent_key: Key from parent level (for recursion)
          sep: Separator to use between nested keys
  
      Returns:
          Flattened dictionary with no nested structures
      """
      items = []
      for k, v in d.items():
          new_key = f"{parent_key}{sep}{k}" if parent_key else k
  
          if isinstance(v, dict):
              # Recursively flatten nested dicts
              items.extend(_flatten_dict(v, new_key, sep=sep).items())
          elif isinstance(v, list):
              # Convert lists to comma-separated strings
              items.append((new_key, str(v)))
          else:
              items.append((new_key, v))
  
      return dict(items)