Core Response Classes

AgentResponse

cortex_agents.core.response.AgentResponse

AgentResponse(raw_response: Any, stream: bool = True, is_async: bool = False)

Bases: BaseSSEResponse

Wrapper for agent run response with easy access to results.

Inherits SSE parsing, iteration, and request_id capture from :class:BaseSSEResponse. Adds Agent-specific convenience properties for text, SQL, thinking, charts, tables, tools, etc.

Examples:

with CortexAgent(...) as agent:
    response = agent.run(
        "What's the revenue?",
        agent_name="MY_AGENT",
        database="MY_DB",
        schema="MY_SCHEMA"
    )

    # Stream events in real-time (response is directly iterable)
    for event in response:
        if event["type"] == "text.delta":
            print(event["data"]["text"], end="", flush=True)

# Or just get final results
print(response.text)        # Final text response
print(response.sql)         # SQL query if any
print(response.thinking)    # Agent reasoning
print(response.query_id)    # Snowflake query ID

# Get SQL execution results
result_set = response.get_sql_result()
if result_set:
    for row in result_set['data']:
        print(row)

# Get structured data
charts = response.get_charts()
tables = response.get_tables()

Initialize response wrapper.

Parameters:

Name	Type	Description	Default
raw_response	Any	Callable that returns httpx streaming context manager, or a dict for non-streaming responses.	required
stream	bool	Whether response is streaming. Defaults to True.	True
is_async	bool	Whether this is an async response. Defaults to False.	False

Source code in cortex_agents/core/response.py

def __init__(self, raw_response: Any, stream: bool = True, is_async: bool = False):
    """Initialize response wrapper.

    Args:
        raw_response: Callable that returns httpx streaming context manager,
            or a dict for non-streaming responses.
        stream: Whether response is streaming. Defaults to True.
        is_async: Whether this is an async response. Defaults to False.
    """
    super().__init__(raw_response, stream=stream, is_async=is_async)

text property

text: str

Get final text response (concatenated from all text events by content_index).

Real SSE streams may emit multiple response.text events at different content_index values (e.g., answer, explanation, methodology). This property concatenates ALL text blocks in content_index order.

Returns:

Name	Type	Description
str	str	Complete text response

thinking property

thinking: str

Get agent's reasoning/thinking tokens (concatenated from all thinking blocks).

Returns:

Name	Type	Description
str	str	Complete thinking process

sql property

sql: str | None

Get SQL query from Cortex Analyst tool (if any).

Returns:

Type	Description
str | None	Optional[str]: SQL query or None

sql_explanation property

sql_explanation: str | None

Get SQL explanation from Cortex Analyst (if any).

Returns:

Type	Description
str | None	Optional[str]: SQL explanation or None

query_id property

query_id: str | None

Get Snowflake query ID from SQL execution (if any).

Useful for tracking query execution in Snowflake.

Returns:

Type	Description
str | None	Optional[str]: Query ID or None

message_id property

message_id: str | None

Get the last assistant's message ID (for thread continuity).

Returns:

Type	Description
str | None	Message ID as provided by the API (typically an integer) or None if not available.

run_id property

run_id: str | None

Get the run ID from metadata events.

The run_id is typically found in the metadata of the final response or in response.metadata events.

Returns:

Type	Description
str | None	Optional[str]: The run ID or None

is_elicitation property

is_elicitation: bool

Check if the response is an elicitation request (asking for more info).

Returns:

Name	Type	Description
bool	bool	True if any text event has is_elicitation set

request_id property

request_id: str | None

Snowflake request ID captured from response headers or event data.

events property

events: list[dict[str, Any]]

All accumulated events (triggers full parse if needed).

get_sql_result

get_sql_result() -> dict | None

Get SQL execution result set from Cortex Analyst (if any).

Returns the full result_set including data, metadata, and statementHandle.

Returns:

Type	Description
dict | None	Optional[Dict]: Result set or None

Examples:

result = response.get_sql_result()
if result:
    data = result['data']  # Array of rows
    metadata = result['resultSetMetaData']
    print(f"Returned {metadata['numRows']} rows")
    for row in data:
        print(row)

Source code in cortex_agents/core/response.py

def get_sql_result(self) -> dict | None:
    """Get SQL execution result set from Cortex Analyst (if any).

    Returns the full result_set including data, metadata, and statementHandle.

    Returns:
        Optional[Dict]: Result set or None

    Examples:
    ```python
    result = response.get_sql_result()
    if result:
        data = result['data']  # Array of rows
        metadata = result['resultSetMetaData']
        print(f"Returned {metadata['numRows']} rows")
        for row in data:
            print(row)
    ```
    """
    self._ensure_parsed()

    # Check tool_result events for result_set
    for event in self._events:
        if event["type"] == "tool_result":
            data = event["data"]
            if data.get("type") == "cortex_analyst_text_to_sql":
                content = data.get("content", [])
                if content and len(content) > 0:
                    json_data = content[0].get("json", {})
                    if "result_set" in json_data:
                        return json_data["result_set"]

    return None

get_charts

get_charts() -> list[dict]

Get all chart specifications (Vega-Lite).

Returns:

Type	Description
list[dict]	list[Dict]: List of chart specs

Source code in cortex_agents/core/response.py

def get_charts(self) -> list[dict]:
    """Get all chart specifications (Vega-Lite).

    Returns:
        list[Dict]: List of chart specs
    """
    self._ensure_parsed()
    charts = []
    for event in self._events:
        if event["type"] == "chart":
            charts.append(event["data"])
    return charts

get_tables

get_tables() -> list[dict]

Get all data tables.

Returns:

Type	Description
list[dict]	list[Dict]: List of tables with result_set data

Source code in cortex_agents/core/response.py

def get_tables(self) -> list[dict]:
    """Get all data tables.

    Returns:
        list[Dict]: List of tables with result_set data
    """
    self._ensure_parsed()
    tables = []
    for event in self._events:
        if event["type"] == "table":
            tables.append(event["data"])
    return tables

get_tool_uses

get_tool_uses() -> list[dict]

Get all tool invocations.

Returns:

Type	Description
list[dict]	list[Dict]: List of tool use events

Source code in cortex_agents/core/response.py

def get_tool_uses(self) -> list[dict]:
    """Get all tool invocations.

    Returns:
        list[Dict]: List of tool use events
    """
    self._ensure_parsed()
    tools = []
    for event in self._events:
        if event["type"] == "tool_use":
            tools.append(event["data"])
    return tools

get_tool_results

get_tool_results() -> list[dict]

Get all tool execution results.

Returns:

Type	Description
list[dict]	list[Dict]: List of tool results

Source code in cortex_agents/core/response.py

def get_tool_results(self) -> list[dict]:
    """Get all tool execution results.

    Returns:
        list[Dict]: List of tool results
    """
    self._ensure_parsed()
    results = []
    for event in self._events:
        if event["type"] == "tool_result":
            results.append(event["data"])
    return results

get_metadata

get_metadata() -> list[dict]

Get message metadata (includes message_id for follow-ups).

Returns:

Type	Description
list[dict]	list[dict]: List of metadata dicts with role and message_id

Source code in cortex_agents/core/response.py

def get_metadata(self) -> list[dict]:
    """Get message metadata (includes message_id for follow-ups).

    Returns:
        list[dict]: List of metadata dicts with role and message_id
    """
    self._ensure_parsed()
    return [event["data"]["metadata"] for event in reversed(self._events) if event["type"] == "metadata"]

get_final_response

get_final_response() -> dict | None

Get the final aggregated response event.

Returns:

Type	Description
dict | None	Optional[Dict]: Complete response or None

Source code in cortex_agents/core/response.py

def get_final_response(self) -> dict | None:
    """Get the final aggregated response event.

    Returns:
        Optional[Dict]: Complete response or None
    """
    self._ensure_parsed()
    for event in reversed(self._events):
        if event["raw_type"] == "response":
            return event["data"]
    return None

get_token_usage

get_token_usage() -> dict | None

Get token usage information from the response metadata.

Returns:

Type	Description
dict | None	Optional[Dict]: Token usage dict (e.g. {"tokens_consumed": 1234}) or None

Source code in cortex_agents/core/response.py

def get_token_usage(self) -> dict | None:
    """Get token usage information from the response metadata.

    Returns:
        Optional[Dict]: Token usage dict (e.g. ``{"tokens_consumed": 1234}``) or None
    """
    self._ensure_parsed()

    # Check final aggregated response
    for event in reversed(self._events):
        if event["raw_type"] == "response":
            metadata = event["data"].get("metadata", {})
            if isinstance(metadata, dict) and "usage" in metadata:
                return metadata["usage"]

    # Check metadata events
    for event in self._events:
        if event["type"] == "metadata":
            usage = event["data"].get("usage")
            if usage:
                return usage

    return None

get_warnings

get_warnings() -> list[str]

Get all warning messages from the response.

Warnings are emitted as response.warning SSE events.

Returns:

Type	Description
list[str]	list[str]: List of warning message strings

Source code in cortex_agents/core/response.py

def get_warnings(self) -> list[str]:
    """Get all warning messages from the response.

    Warnings are emitted as ``response.warning`` SSE events.

    Returns:
        list[str]: List of warning message strings
    """
    self._ensure_parsed()
    warnings = []
    for event in self._events:
        if event["raw_type"] in ("response.warning", "warning"):
            msg = event["data"].get("message") or event["data"].get("warning", "")
            if msg:
                warnings.append(msg)
    return warnings

get_suggested_queries

get_suggested_queries() -> list[str]

Get suggested follow-up queries from the response.

These are emitted as response.suggested_queries SSE events.

Returns:

Type	Description
list[str]	list[str]: List of suggested query strings

Source code in cortex_agents/core/response.py

def get_suggested_queries(self) -> list[str]:
    """Get suggested follow-up queries from the response.

    These are emitted as ``response.suggested_queries`` SSE events.

    Returns:
        list[str]: List of suggested query strings
    """
    self._ensure_parsed()
    queries: list[str] = []
    for event in self._events:
        if event["raw_type"] == "response.suggested_queries":
            data = event["data"]
            if isinstance(data, dict):
                sq = data.get("suggested_queries", [])
                if isinstance(sq, list):
                    queries.extend(sq)
                elif isinstance(sq, str):
                    queries.append(sq)
    return queries

get_annotations

get_annotations() -> list[dict]

Get all annotations from text events.

Annotations (e.g., citations, references) may appear in response.text or response.text.annotation events.

Returns:

Type	Description
list[dict]	list[Dict]: List of annotation dicts

Source code in cortex_agents/core/response.py

def get_annotations(self) -> list[dict]:
    """Get all annotations from text events.

    Annotations (e.g., citations, references) may appear in
    ``response.text`` or ``response.text.annotation`` events.

    Returns:
        list[Dict]: List of annotation dicts
    """
    self._ensure_parsed()
    annotations: list[dict] = []
    for event in self._events:
        if event["raw_type"] == "response.text.annotation":
            annotations.append(event["data"])
        elif event["type"] == "text":
            ann = event["data"].get("annotations")
            if isinstance(ann, list):
                annotations.extend(ann)
    return annotations

stream

stream() -> Generator[dict[str, Any], None, None]

Backward-compatible alias for iter(self).

Source code in cortex_agents/_base_response.py

def stream(self) -> Generator[dict[str, Any], None, None]:
    """Backward-compatible alias for ``iter(self)``."""
    return self._stream_events()

astream async

astream() -> AsyncGenerator[dict[str, Any], None]

Backward-compatible alias for async for … in self.

Source code in cortex_agents/_base_response.py

async def astream(self) -> AsyncGenerator[dict[str, Any], None]:
    """Backward-compatible alias for ``async for … in self``."""
    async for event in self._astream_events():
        yield event

EventType

cortex_agents.core.response.EventType

Bases: str, Enum

Enumeration of SSE event types from agent:run endpoint.

SSE (Server-Sent Events) events include various message types for streaming agent responses including text deltas, tool usage, metadata, and errors.

AnalystResponse

cortex_agents._analyst_response.AnalystResponse

AnalystResponse(response: Any, stream: bool = True, is_async: bool = False, request_messages: list[dict[str, Any]] | None = None)

Bases: BaseSSEResponse

Wrapper for Cortex Analyst responses with convenient property access.

Provides easy access to generated SQL statements, interpretation text, suggestions for ambiguous questions, confidence information, and streaming events.

Examples:

response = analyst.message(
    "What is the revenue?",
    semantic_model_file="@stage/model.yaml"
)

# Stream events in real-time (response is directly iterable)
for event in response:
    if event["type"] == "text.delta":
        print(event["data"]["text"], end="")

# Or access final properties
print(response.text)        # Interpretation
print(response.sql)         # Generated SQL
print(response.suggestions) # If ambiguous

Initialize the response wrapper.

Parameters:

Name	Type	Description	Default
response	Any	Callable that returns httpx streaming context manager, or a dict for non-streaming responses.	required
stream	bool	Whether this is a streaming response. Defaults to True.	True
is_async	bool	Whether this is an async response. Defaults to False.	False
request_messages	list[dict[str, Any]] | None	Original request messages for conversation continuity.	None

Source code in cortex_agents/_analyst_response.py

def __init__(
    self,
    response: Any,
    stream: bool = True,
    is_async: bool = False,
    request_messages: list[dict[str, Any]] | None = None,
) -> None:
    """Initialize the response wrapper.

    Args:
        response: Callable that returns httpx streaming context manager,
            or a dict for non-streaming responses.
        stream: Whether this is a streaming response. Defaults to True.
        is_async: Whether this is an async response. Defaults to False.
        request_messages: Original request messages for conversation continuity.
    """
    super().__init__(response, stream=stream, is_async=is_async)
    self._parsed_data: dict[str, Any] | None = None
    self._request_messages: list[dict[str, Any]] | None = (
        deepcopy(request_messages) if request_messages is not None else None
    )

text property

text: str

Get the interpretation text from the analyst.

sql property

sql: str | None

Get the generated SQL statement (None if question was ambiguous).

suggestions property

suggestions: list[str]

Get suggested questions (if question was ambiguous).

confidence property

confidence: dict | None

Get confidence information about the SQL generation.

verified_query_used property

verified_query_used: dict | None

Get verified query information if one was used.

query_id property

query_id: str | None

Get the query ID from SQL execution.

request_id property

request_id: str | None

Get the Snowflake request ID for this response.

Checks HTTP response headers first (via BaseSSEResponse), then falls back to the parsed response data.

warnings property

warnings: list[dict]

Get any warnings from the response.

response_metadata property

response_metadata: dict

Get response metadata (models used, question category, etc.).

conversation_messages property

conversation_messages: list[dict[str, Any]]

Get the full messages array for multi-turn conversations.

Examples:

response1 = analyst.message("First question", ...)
response2 = analyst.message(
    "Follow-up question",
    messages=response1.conversation_messages,
    ...
)

sql_explanation property

sql_explanation: str | None

Get Analyst's explanation of the generated SQL.

result_set property

result_set: dict | None

Get the SQL execution result set (if available).

semantic_model_selection property

semantic_model_selection: str | None

Get which semantic model was selected (when multiple provided).

cortex_search_retrieval property

cortex_search_retrieval: list[dict[str, Any]]

Get Cortex Search retrieval results.

is_semantic_sql property

is_semantic_sql: bool

Check if generated SQL uses Semantic Views.

analyst_latency_ms property

analyst_latency_ms: float | None

Get analyst processing latency in milliseconds.

question_category property

question_category: str | None

Get the question category determined by the analyst.

model_names property

model_names: list[str]

Get the names of LLM models used for processing.

analyst_orchestration_path property

analyst_orchestration_path: str | None

Get the orchestration path taken during analysis.

events property

events: list[dict[str, Any]]

All accumulated events (triggers full parse if needed).

stream

stream() -> Generator[dict[str, Any], None, None]

Backward-compatible alias for iter(self).

Source code in cortex_agents/_base_response.py

def stream(self) -> Generator[dict[str, Any], None, None]:
    """Backward-compatible alias for ``iter(self)``."""
    return self._stream_events()

astream async

astream() -> AsyncGenerator[dict[str, Any], None]

Backward-compatible alias for async for … in self.

Source code in cortex_agents/_base_response.py

async def astream(self) -> AsyncGenerator[dict[str, Any], None]:
    """Backward-compatible alias for ``async for … in self``."""
    async for event in self._astream_events():
        yield event