@@ -39,6 +39,18 @@ Changelog
   attempt is checked and counted on its own, unchanged. See :ref:`the cross-attempt usage
   budget <agent-usage-budget>`.
 
+.. note::
+  ``get_schema`` on ``SQLToolset`` and ``DataFusionToolset`` now returns a JSON object
+  ``{"columns": [{"name", "type"}, ...], "column_count": N}`` instead of a bare JSON array of
+  columns. The tool also accepts an optional ``name_contains`` substring filter, and on a table
+  with more columns than ``max_columns`` (default 100), or one whose serialized columns exceed
+  ``max_result_bytes``, it returns a bounded summary (``column_count``, a ``type_histogram`` and a
+  ``sample_columns`` preview, with ``truncated``, ``truncated_by`` and a ``hint``) in place of the
+  full list. Update any system prompt or direct ``call_tool("get_schema", ...)`` caller that read
+  the old top-level array: check ``truncated`` first, then read ``result["columns"]`` on a full
+  result or ``result["sample_columns"]`` on a summary. A summary carries no ``columns`` key, so
+  ``result["columns"]`` raises ``KeyError`` on any table wide enough to be summarized.
+
 0.10.0
 ......
 
 
@@ -37,7 +37,9 @@ querying files on object stores (S3, GCS, local filesystem, Iceberg) via Apache
    * - ``list_tables``
      - Lists registered table names
    * - ``get_schema``
-     - Returns column names and types for a table (Arrow schema)
+     - Returns a table's columns (Arrow schema) as JSON, with a ``name_contains``
+       filter and a bounded summary on very wide tables (see
+       :ref:`bounded-schema-results`)
    * - ``query``
      - Executes a SQL query and returns bounded, columnar JSON (see
        :ref:`bounded-query-results`)
@@ -84,8 +86,11 @@ Parameters
   permitted. DataFusion on object stores is mostly read-only, but it does
   support DDL for in-memory tables; this guard blocks those by default.
 - ``max_rows``: Maximum rows returned from the ``query`` tool. Default ``50``.
-- ``max_result_bytes``: Budget for the serialized ``query`` result. Default 64 KiB.
-  See :ref:`bounded-query-results`.
+- ``max_result_bytes``: Budget for the serialized ``query`` result, and the byte backstop
+  that also triggers the ``get_schema`` summary. Default 64 KiB.
+  See :ref:`bounded-query-results` and :ref:`bounded-schema-results`.
+- ``max_columns``: Maximum columns ``get_schema`` returns in full. Default ``100``.
+  Above it the result becomes a bounded summary. See :ref:`bounded-schema-results`.
 - ``max_retries``: How many times the model may correct a failed call to these
   tools. Default ``None``, the agent's ``retries``. See :ref:`toolset-retry-budget`.
 
 
@@ -30,7 +30,8 @@ Curated toolset wrapping
    * - ``list_tables``
      - Lists available table names (filtered by ``allowed_tables`` if set)
    * - ``get_schema``
-     - Returns column names and types for a table
+     - Returns a table's columns as JSON, with a ``name_contains`` filter and a
+       bounded summary on very wide tables (see :ref:`bounded-schema-results`)
    * - ``query``
      - Executes a SQL query and returns bounded, columnar JSON (see
        :ref:`bounded-query-results`)
@@ -203,8 +204,11 @@ Parameters
 - ``max_rows``: Maximum rows returned from the ``query`` tool. Default ``50``.
   Rows beyond it are not read out of a DBAPI cursor; what the driver has already
   transferred is its own call. See :ref:`bounded-query-results`.
-- ``max_result_bytes``: Budget for the serialized ``query`` result. Default 64 KiB.
-  See :ref:`bounded-query-results`.
+- ``max_result_bytes``: Budget for the serialized ``query`` result, and the byte backstop
+  that also triggers the ``get_schema`` summary. Default 64 KiB.
+  See :ref:`bounded-query-results` and :ref:`bounded-schema-results`.
+- ``max_columns``: Maximum columns ``get_schema`` returns in full. Default ``100``.
+  Above it the result becomes a bounded summary. See :ref:`bounded-schema-results`.
 - ``max_retries``: How many times the model may correct a failed call to these
   tools. Default ``None``, the agent's ``retries``. See :ref:`toolset-retry-budget`.
 
@@ -263,6 +267,37 @@ result several-fold, so results that fit before still fit. Lower ``max_result_by
 when an agent makes many queries in one run, since every result is re-paid on every
 later request.
 
+.. _bounded-schema-results:
+
+Bounded schema results
+----------------------
+
+``get_schema`` has the same context problem as ``query`` but cannot be solved the same
+way. Column names are what the agent needs to write SQL, so truncating to the first N
+columns would leave it unable to reference or discover the rest. The tool filters and
+summarizes instead.
+
+**Filter with** ``name_contains``. Pass a case-insensitive substring to get back only
+the columns whose name contains it, so on a very wide table the agent asks for the
+columns relevant to its question rather than all of them.
+
+**A summary replaces a very wide list.** Above ``max_columns`` (default ``100``), or when
+the serialized columns exceed ``max_result_bytes``, the full list is replaced by a
+summary that reports the shape and points at the filter:
+
+.. code-block:: json
+
+    {"column_count": 3200, "truncated": true, "truncated_by": "max_columns",
+     "hint": "...", "type_histogram": {"VARCHAR": 2000, "NUMBER": 1200},
+     "sample_columns": [{"name": "id", "type": "NUMBER"}]}
+
+``truncated_by`` is ``max_columns`` or ``max_result_bytes``. ``type_histogram`` counts
+columns per type (capped to the most common, the tail folded into one entry), and
+``sample_columns`` previews the first columns -- both only while they fit the budget, so
+a pathologically small budget still returns the ``column_count`` and ``hint``. A filtered
+call echoes ``name_contains`` and adds ``total_columns`` so a subset is never mistaken for
+the whole table.
+
 When to choose it
 -----------------
 
 
@@ -37,9 +37,12 @@
 
 from airflow.providers.common.ai.utils.masking import dumps_masked
 from airflow.providers.common.ai.utils.query_results import (
+    DEFAULT_MAX_COLUMNS,
     DEFAULT_MAX_RESULT_BYTES,
+    GET_SCHEMA_TOOL_DESCRIPTION as _GET_SCHEMA_DESCRIPTION,
     QUERY_TOOL_DESCRIPTION as _QUERY_DESCRIPTION,
     build_query_result,
+    build_schema_result,
 )
 from airflow.providers.common.ai.utils.tool_definition import build_args_validator
 from airflow.providers.common.ai.utils.toolset_base import AirflowToolset, validate_max_retries
@@ -61,6 +64,13 @@
     "type": "object",
     "properties": {
         "table_name": {"type": "string", "description": "Name of the table to inspect."},
+        "name_contains": {
+            "type": "string",
+            "description": (
+                "Return only columns whose name contains this substring (case-insensitive). "
+                "Use it to find the relevant columns on a very wide table."
+            ),
+        },
     },
     "required": ["table_name"],
 }
@@ -112,7 +122,8 @@ class DataFusionToolset(AirflowToolset):
     :param max_rows: Maximum number of rows returned from the ``query`` tool.
         Default ``50``. The query is limited to ``max_rows + 1`` rows, so a large
         result is never fully materialized; the extra row only signals truncation.
-    :param max_result_bytes: Budget for the serialized ``query`` result, in bytes.
+    :param max_result_bytes: Budget for the serialized ``query`` result, in bytes, and the
+        byte backstop that also triggers the ``get_schema`` summary (see ``max_columns``).
         Default 64 KiB. ``max_rows`` bounds rows, which says nothing about size: one
         row of a 3000-column table is larger than a thousand rows of a narrow one, and
         a tool result stays in the model's message history for the rest of the run, so
@@ -121,6 +132,11 @@ class DataFusionToolset(AirflowToolset):
         rather than skipping it and packing later ones, so one wide row early in the
         result ends it. The result reports which limit it hit so the agent can narrow
         its projection rather than page through the table.
+    :param max_columns: Maximum number of columns ``get_schema`` returns in full. Default
+        ``100``. Above it -- or when the serialized columns exceed ``max_result_bytes`` --
+        the full list is replaced by a bounded summary (column count, a type histogram, a
+        sample of columns) that points the agent at the ``name_contains`` filter, so a
+        several-thousand-column table cannot exhaust the context before a query is written.
     :param max_retries: How many times the model may correct a failed call to one of these
         tools before the run fails. ``None`` (the default) uses the agent's tool retry
         budget, its ``retries``, as pydantic-ai's own toolsets do.
@@ -133,6 +149,7 @@ def __init__(
         allow_writes: bool = False,
         max_rows: int = 50,
         max_result_bytes: int = DEFAULT_MAX_RESULT_BYTES,
+        max_columns: int = DEFAULT_MAX_COLUMNS,
         max_retries: int | None = None,
     ) -> None:
         self._max_retries = validate_max_retries(max_retries)
@@ -142,6 +159,7 @@ def __init__(
         self._allow_writes = allow_writes
         self._max_rows = max_rows
         self._max_result_bytes = max_result_bytes
+        self._max_columns = max_columns
         self._engine: DataFusionEngine | None = None
 
     @property
@@ -164,7 +182,7 @@ async def get_tools(self, ctx: RunContext[Any]) -> dict[str, ToolsetTool[Any]]:
 
         for name, description, schema in (
             ("list_tables", "List available table names.", _LIST_TABLES_SCHEMA),
-            ("get_schema", "Get column names and types for a table.", _GET_SCHEMA_SCHEMA),
+            ("get_schema", _GET_SCHEMA_DESCRIPTION, _GET_SCHEMA_SCHEMA),
             ("query", _QUERY_DESCRIPTION, _QUERY_SCHEMA),
         ):
             tool_def = ToolDefinition(
@@ -192,7 +210,9 @@ async def execute_tool(
         if name == "list_tables":
             return await self.run_blocking(self._list_tables)
         if name == "get_schema":
-            return await self.run_blocking(self._get_schema, tool_args["table_name"])
+            return await self.run_blocking(
+                self._get_schema, tool_args["table_name"], tool_args.get("name_contains")
+            )
         if name == "query":
             return await self.run_blocking(self._query, tool_args["sql"])
         raise ValueError(f"Unknown tool: {name!r}")
@@ -206,7 +226,7 @@ def _list_tables(self) -> str:
             log.warning("list_tables failed: %s", ex)
             return dumps_masked({"error": str(ex)})
 
-    def _get_schema(self, table_name: str) -> str:
+    def _get_schema(self, table_name: str, name_contains: str | None = None) -> str:
         engine = self._get_engine()
         # session_context lookup is required here instead of engine.registered_tables,
         # because registered_tables only tracks tables registered via datasource config.
@@ -220,7 +240,12 @@ def _get_schema(self, table_name: str) -> str:
         # TODO: refactor engine.get_schema() to return JSON and update this accordingly
         table = engine.session_context.table(table_name)
         columns = [{"name": f.name, "type": str(f.type)} for f in table.schema()]
-        return dumps_masked(columns)
+        return build_schema_result(
+            columns,
+            max_columns=self._max_columns,
+            max_result_bytes=self._max_result_bytes,
+            name_contains=name_contains,
+        )
 
     def _query(self, sql: str) -> str:
         try:
 
@@ -42,9 +42,12 @@
 
 from airflow.providers.common.ai.utils.masking import dumps_masked
 from airflow.providers.common.ai.utils.query_results import (
+    DEFAULT_MAX_COLUMNS,
     DEFAULT_MAX_RESULT_BYTES,
+    GET_SCHEMA_TOOL_DESCRIPTION as _GET_SCHEMA_DESCRIPTION,
     QUERY_TOOL_DESCRIPTION as _QUERY_DESCRIPTION,
     build_query_result,
+    build_schema_result,
 )
 from airflow.providers.common.ai.utils.tool_definition import build_args_validator, return_schema_kwargs
 from airflow.providers.common.ai.utils.toolset_base import AirflowToolset, validate_max_retries
@@ -73,6 +76,13 @@
     "type": "object",
     "properties": {
         "table_name": {"type": "string", "description": "Name of the table to inspect."},
+        "name_contains": {
+            "type": "string",
+            "description": (
+                "Return only columns whose name contains this substring (case-insensitive). "
+                "Use it to find the relevant columns on a very wide table."
+            ),
+        },
     },
     "required": ["table_name"],
 }
@@ -241,7 +251,8 @@ class SQLToolset(AirflowToolset):
         the time the first row is read, so only the per-row Python conversion is
         skipped. Treat this as a bound on what the agent is shown, not as a guarantee
         that ``SELECT * FROM huge_table`` is cheap.
-    :param max_result_bytes: Budget for the serialized ``query`` result, in bytes.
+    :param max_result_bytes: Budget for the serialized ``query`` result, in bytes, and the
+        byte backstop that also triggers the ``get_schema`` summary (see ``max_columns``).
         Default 64 KiB. ``max_rows`` bounds rows, which says nothing about size: one
         row of a 3000-column table is larger than a thousand rows of a narrow one, and
         a tool result stays in the model's message history for the rest of the run, so
@@ -250,6 +261,11 @@ class SQLToolset(AirflowToolset):
         rather than skipping it and packing later ones, so one wide row early in the
         result ends it. The result reports which limit it hit so the agent can narrow
         its projection rather than page through the table.
+    :param max_columns: Maximum number of columns ``get_schema`` returns in full. Default
+        ``100``. Above it -- or when the serialized columns exceed ``max_result_bytes`` --
+        the full list is replaced by a bounded summary (column count, a type histogram, a
+        sample of columns) that points the agent at the ``name_contains`` filter, so a
+        several-thousand-column table cannot exhaust the context before a query is written.
     :param max_retries: How many times the model may correct a failed call to one of these
         tools before the run fails. ``None`` (the default) uses the agent's tool retry
         budget, its ``retries``, as pydantic-ai's own toolsets do.
@@ -269,6 +285,7 @@ def __init__(
         allow_writes: bool = False,
         max_rows: int = 50,
         max_result_bytes: int = DEFAULT_MAX_RESULT_BYTES,
+        max_columns: int = DEFAULT_MAX_COLUMNS,
         max_retries: int | None = None,
     ) -> None:
         self._max_retries = validate_max_retries(max_retries)
@@ -294,6 +311,7 @@ def __init__(
         self._allow_writes = allow_writes
         self._max_rows = max_rows
         self._max_result_bytes = max_result_bytes
+        self._max_columns = max_columns
         self._hook: DbApiHook | None = None
 
         # Canonical ``(catalog, schema, table)`` view of allowed_tables for membership
@@ -379,7 +397,7 @@ async def get_tools(self, ctx: RunContext[Any]) -> dict[str, ToolsetTool[Any]]:
 
         for name, description, schema in (
             ("list_tables", "List available table names in the database.", _LIST_TABLES_SCHEMA),
-            ("get_schema", "Get column names and types for a table.", _GET_SCHEMA_SCHEMA),
+            ("get_schema", _GET_SCHEMA_DESCRIPTION, _GET_SCHEMA_SCHEMA),
             ("query", _QUERY_DESCRIPTION, _QUERY_SCHEMA),
             ("check_query", "Validate SQL syntax without executing it.", _CHECK_QUERY_SCHEMA),
         ):
@@ -431,7 +449,7 @@ def _run_tool(self, name: str, tool_args: dict[str, Any]) -> str:
         if name == "list_tables":
             return self._list_tables()
         if name == "get_schema":
-            return self._get_schema(tool_args["table_name"])
+            return self._get_schema(tool_args["table_name"], tool_args.get("name_contains"))
         if name == "query":
             return self._query(tool_args["sql"])
         return self._check_query(tool_args["sql"])
@@ -475,13 +493,18 @@ def add(schema: str | None, name: str, display: str) -> None:
 
         return dumps_masked(tables)
 
-    def _get_schema(self, table_name: str) -> str:
+    def _get_schema(self, table_name: str, name_contains: str | None = None) -> str:
         schema, table = self._split_table_identifier(table_name)
         if not self._is_ref_allowed("", schema, table):
             return dumps_masked({"error": f"Table {table_name!r} is not in the allowed tables list."})
         hook = self._get_db_hook()
         columns = hook.get_table_schema(table, schema=schema)
-        return dumps_masked(columns)
+        return build_schema_result(
+            columns,
+            max_columns=self._max_columns,
+            max_result_bytes=self._max_result_bytes,
+            name_contains=name_contains,
+        )
 
     def _dialect_for_validation(self) -> str | None:
         """Resolve the hook's sqlglot dialect so DESCRIBE/SHOW validate correctly."""