Polygon-io MCP Server

Instructions

Implementation Reference

• src/mcp_massive/server.py:32-64 (handler)

  The get_aggs tool handler: decorated with @poly_mcp.tool for registration, defines input parameters (serving as schema), calls massive_client.get_aggs API, formats response to CSV via json_to_csv helper, handles errors.
  @poly_mcp.tool(annotations=ToolAnnotations(readOnlyHint=True)) async def get_aggs( ticker: str, multiplier: int, timespan: str, from_: Union[str, int, datetime, date], to: Union[str, int, datetime, date], adjusted: Optional[bool] = None, sort: Optional[str] = None, limit: Optional[int] = 10, params: Optional[Dict[str, Any]] = None, ) -> str: """ List aggregate bars for a ticker over a given date range in custom time window sizes. """ try: results = massive_client.get_aggs( ticker=ticker, multiplier=multiplier, timespan=timespan, from_=from_, to=to, adjusted=adjusted, sort=sort, limit=limit, params=params, raw=True, ) # Parse the binary data to string and then to JSON return json_to_csv(results.data.decode("utf-8")) except Exception as e: return f"Error: {e}"
• src/mcp_massive/formatters.py:7-105 (helper)

  json_to_csv helper function imported and used in get_aggs to convert API JSON response to flattened CSV string for tool output. Includes _flatten_dict sub-helper.
  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)
• src/mcp_massive/server.py:32-32 (registration)

  @poly_mcp.tool decorator registers get_aggs as an MCP tool with readOnlyHint annotation.
  @poly_mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
• src/mcp_massive/server.py:33-43 (schema)

  Function signature with type hints defines the input schema and docstring for the get_aggs tool, used by FastMCP for tool schema generation.
  async def get_aggs( ticker: str, multiplier: int, timespan: str, from_: Union[str, int, datetime, date], to: Union[str, int, datetime, date], adjusted: Optional[bool] = None, sort: Optional[str] = None, limit: Optional[int] = 10, params: Optional[Dict[str, Any]] = None, ) -> str: