Zaturn

Instructions

Implementation Reference

• zaturn/tools/visualizations.py:123-157 (handler)

  The main handler function for the 'histogram' tool. It takes source_id, query, column, optional color and nbins; executes the query, generates a histogram with plotly.express, converts to PNG image, returns ImageContent or error string.
  def histogram(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') ], column: Annotated[ str, Field(description='Column name from SQL result to use for the histogram') ], color: Annotated[ str | None, Field(description='Optional; column name from SQL result to use for drawing multiple colored histograms representing another dimension') ] = None, nbins: Annotated[ int | None, Field(description='Optional; number of bins') ] = None, ) -> str | ImageContent: """ Run query against specified source and make a histogram 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.histogram(df, x=column, color=color, nbins=nbins) fig.update_xaxes(autotickangles=[0, 45, 60, 90]) return _fig_to_image(fig) except Exception as e: return str(e)
• zaturn/tools/visualizations.py:123-140 (schema)

  Pydantic Field descriptions and types defining the input schema for the histogram tool.
  def histogram(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') ], column: Annotated[ str, Field(description='Column name from SQL result to use for the histogram') ], color: Annotated[ str | None, Field(description='Optional; column name from SQL result to use for drawing multiple colored histograms representing another dimension') ] = None, nbins: Annotated[ int | None, Field(description='Optional; number of bins') ] = None, ) -> str | ImageContent: """
• zaturn/mcp/__init__.py:89-95 (registration)

  Final MCP registration: creates ZaturnTools instance and registers all its tools (including histogram) to FastMCP server.
  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/visualizations.py:27-40 (registration)

  Intermediate registration: adds histogram method to Visualizations.tools list.
  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/__init__.py:1-13 (registration)

  Aggregates tools from Core and Visualizations (including histogram) into ZaturnTools.tools.
  from zaturn.tools import core, visualizations class ZaturnTools: def __init__(self, data_sources): self.tools = [ *core.Core(data_sources).tools, *visualizations.Visualizations(data_sources).tools, ]
• zaturn/tools/visualizations.py:13-22 (helper)

  Helper function to convert Plotly figure to base64-encoded ImageContent for all visualization tools.
  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, )