Zaturn

Instructions

Implementation Reference

• zaturn/tools/visualizations.py:51-84 (handler)

  The core handler function for the 'scatter_plot' tool. It takes source_id, SQL query, x/y columns, optional color column; fetches data via _get_df_from_source, creates plotly scatter plot, converts to ImageContent PNG, or returns error string.
  def scatter_plot(self, source_id: Annotated[ str, Field(description='The data source to run the query on') ], query: Annotated[ str, Field(description='SQL query to run on the data source') ], x: Annotated[ str, Field(description='Column name from SQL result to use for x-axis') ], y: Annotated[ str, Field(description='Column name from SQL result to use for y-axis') ], color: Annotated[ str | None, Field(description='Optional; column name from SQL result to use for coloring the points, with color representing another dimension') ] = None, ) -> str | ImageContent: """ Run query against specified source and make a scatter plot using result For both csv and parquet sources, use DuckDB SQL syntax Use 'CSV' as the table name in the SQL query for csv sources. Use 'PARQUET' as the table name in the SQL query for parquet sources. This will return an image of the plot """ try: df = self._get_df_from_source(source_id, query) fig = px.scatter(df, x=x, y=y, color=color) fig.update_xaxes(autotickangles=[0, 45, 60, 90]) return _fig_to_image(fig) except Exception as e: return str(e)
• zaturn/tools/visualizations.py:51-67 (schema)

  Pydantic-style input schema defined via Annotated and Field for source_id (str), query (str), x (str), y (str), color (str|None). Output: str | ImageContent.
  def scatter_plot(self, source_id: Annotated[ str, Field(description='The data source to run the query on') ], query: Annotated[ str, Field(description='SQL query to run on the data source') ], x: Annotated[ str, Field(description='Column name from SQL result to use for x-axis') ], y: Annotated[ str, Field(description='Column name from SQL result to use for y-axis') ], color: Annotated[ str | None, Field(description='Optional; column name from SQL result to use for coloring the points, with color representing another dimension') ] = None, ) -> str | ImageContent:
• zaturn/mcp/__init__.py:89-95 (registration)

  Final MCP registration: Instantiates ZaturnTools with data sources, creates FastMCP server, loops over zaturn_tools.tools (including scatter_plot) and registers each as a Tool.
  def ZaturnMCP(sources): zaturn_tools = ZaturnTools(sources) zaturn_mcp = FastMCP() for tool_function in zaturn_tools.tools: zaturn_mcp.add_tool(Tool.from_function(tool_function)) return zaturn_mcp
• zaturn/tools/__init__.py:4-13 (registration)

  ZaturnTools aggregates tools from core and visualizations.Visualizations (which includes scatter_plot), unpacking their self.tools lists.
  class ZaturnTools: def __init__(self, data_sources): self.tools = [ *core.Core(data_sources).tools, *visualizations.Visualizations(data_sources).tools, ]
• zaturn/tools/visualizations.py:27-40 (registration)

  Visualizations.__init__ adds scatter_plot (and other viz tools) to self.tools list, which propagates to ZaturnTools and MCP registration.
  def __init__(self, data_sources): self.data_sources = data_sources self.tools = [ self.scatter_plot, self.line_plot, self.histogram, self.strip_plot, self.box_plot, self.bar_plot, self.density_heatmap, self.polar_scatter, self.polar_line, ]
• zaturn/tools/visualizations.py:13-22 (helper)

  Helper to convert Plotly figure to MCP ImageContent with base64 PNG data.
  def _fig_to_image(fig): fig_encoded = b64encode(fig.to_image(format='png')).decode() img_b64 = "data:image/png;base64," + fig_encoded return ImageContent( type = 'image', data = fig_encoded, mimeType = 'image/png', annotations = None, )