def make_duckdb_tool()

in databao/duckdb/react_tools.py [0:0]

• 21 lines of code
• 2 McCabe index (conditional complexity)

def make_duckdb_tool(con: DuckDBPyConnection) -> Any:
    """
    Create a DuckDB SQL execution tool for LangChain executors.

    Args:
        con: DuckDB connection to execute queries against.

    Returns:
        A LangChain tool that executes SQL queries.
    """

    @tool("execute_sql")
    def execute_sql(sql: str, limit: int = 10) -> str:
        """
        Execute any SQL against DuckDB.

        Args:
            sql: The SQL statement to execute (single statement).
            limit: Optional row cap for result-returning statements (10 by default).

        Returns:
            JSON string: { "columns": [...], "rows": str, "limit": int, "note": str }
        """
        try:
            df = execute_duckdb_sql(sql, con, limit=limit)
            payload = {
                "columns": list(df.columns),
                "rows": df.to_string(index=False),
                "limit": limit,
                "note": "Query executed successfully",
            }
            return json.dumps(payload)
        except Exception as e:
            payload = {
                "columns": [],
                "rows": [],
                "limit": limit,
                "note": f"SQL error: {type(e).__name__}: {e}",
            }
            return json.dumps(payload)

    return execute_sql