Canvas Agents: How Data, Chart, and Python Agents Split the Work

In AskTable Canvas, Data nodes produce SQL, Chart nodes build visual configs, and Python nodes run transforms. This post sketches how we split responsibilities across agents.

---

Agent roles

DataNodeAgent — SQL

• •Parse the question
• •Retrieve metadata (schema linking)
• •Generate and execute SQL
• •Return tables and short explanations

Tools: schema_linker, sql_executor, sql_guard

ChartNodeAgent — charts

• •Inspect parent data
• •Pick chart type
• •Emit ECharts-style config

Tools: chart_recommender, echarts_generator

PythonNodeAgent — code

• •Translate the ask into Python
• •Execute safely and return frames or metrics

Tools: python_executor, dataframe_loader

---

Base agent

class BaseAgent(ABC):
    """Base agent for a Canvas node."""

    def __init__(self, node: NodeModel):
        self.node = node
        self.messages: list[dict] = []
        self.result: dict = {}

    @abstractmethod
    async def run(self, message: str) -> AsyncGenerator[dict, None]:
        """Stream events."""
        ...

    @abstractmethod
    def get_result(self) -> dict:
        ...

    def get_messages(self) -> list[dict]:
        return self.messages

---

DataNodeAgent (illustrative)

class DataNodeAgent(BaseAgent):
    async def run(self, message: str) -> AsyncGenerator[dict, None]:
        yield {"type": "status", "data": "Linking schema..."}
        meta_context = await self.schema_linker.link(message)

        yield {"type": "status", "data": "Generating SQL..."}
        sql = await self.generate_sql(message, meta_context)
        yield {"type": "sql", "data": sql}

        yield {"type": "status", "data": "Executing..."}
        df = await self.datasource.execute_sql(sql)
        yield {"type": "dataframe", "data": df.to_dict()}

        yield {"type": "status", "data": "Summarizing..."}
        explanation = await self.generate_explanation(message, df)
        yield {"type": "explanation", "data": explanation}

        self.result = {"sql": sql, "dataframe": df, "explanation": explanation}

---

ChartNodeAgent (illustrative)

class ChartNodeAgent(BaseAgent):
    async def run(self, message: str) -> AsyncGenerator[dict, None]:
        yield {"type": "status", "data": "Profiling data..."}
        df = self.parent_contexts[0]["dataframe"]

        yield {"type": "status", "data": "Choosing chart..."}
        chart_type = await self.recommend_chart_type(df, message)
        yield {"type": "chart_type", "data": chart_type}

        yield {"type": "status", "data": "Building config..."}
        config = await self.generate_chart_config(df, chart_type, message)
        yield {"type": "config", "data": config}

        self.result = {"chart_type": chart_type, "config": config}

---

Tool registry

class ToolRegistry:
    def __init__(self):
        self.tools: dict[str, Callable] = {}

    def register(self, name: str, func: Callable):
        self.tools[name] = func

    async def call(self, name: str, **kwargs) -> Any:
        if name not in self.tools:
            raise ValueError(f"Tool {name} not found")
        return await self.tools[name](**kwargs)

registry = ToolRegistry()
registry.register("execute_sql", execute_sql)
registry.register("generate_chart", generate_chart)

---

Takeaways

• •Clear roles per node type
• •Shared tools through a registry
• •Streaming status for responsive UIs
• •Easy to extend with new node kinds

See also: Canvas streaming execution · asktable.com