Cortex Analyst

cortex_agents.analyst.CortexAnalyst

CortexAnalyst(account_url: str | None = None, pat: str | None = None, enable_logging: bool = True, token_type: str | None = None)

Bases: BaseAgent

Client for Snowflake Cortex Analyst.

Provides a simple interface for generating SQL from natural language questions using semantic models or semantic views. Supports multi-turn conversations, streaming responses, and user feedback submission.

Examples:

from cortex_agents import CortexAnalyst

# Use with context manager for automatic cleanup
with CortexAnalyst() as analyst:
    response = analyst.message(
        "Which company had the most revenue?",
        semantic_model_file="@my_stage/model.yaml"
    )

Initialize the Cortex Analyst client.

Source code in cortex_agents/analyst.py

def __init__(
    self,
    account_url: str | None = None,
    pat: str | None = None,
    enable_logging: bool = True,
    token_type: str | None = None,
) -> None:
    """Initialize the Cortex Analyst client."""
    super().__init__(account_url=account_url, pat=pat, enable_logging=enable_logging, token_type=token_type)
    self.session: httpx.Client = httpx.Client(
        headers=self._build_headers(),
        timeout=httpx.Timeout(connect=30.0, read=900.0, write=30.0, pool=30.0),
    )
    self._transport = SyncTransport(
        session=self.session,
        build_url=self._get_url,
        log_request=self._log_request,
        log_response=self._log_response,
        logger=logger,
    )
    if enable_logging:
        logger.info("CortexAnalyst initialized")

close

close() -> None

Close the session and cleanup resources.

Source code in cortex_agents/analyst.py

def close(self) -> None:
    """Close the session and cleanup resources."""
    if self.session:
        self.session.close()
        logger.debug("CortexAnalyst session closed")

message

message(question: str, semantic_model_file: str | None = None, semantic_model: str | None = None, semantic_view: str | None = None, semantic_models: list[dict[str, Any]] | None = None, messages: list[dict[str, Any]] | None = None, stream: bool = True) -> AnalystResponse

Send a message to Cortex Analyst to generate SQL from natural language.

You must specify ONE of: semantic_model_file, semantic_model, semantic_view, or semantic_models.

Parameters:

Name	Type	Description	Default
question	str	The natural language question (str)	required
semantic_model_file	str | None	Path to semantic model YAML file on stage (e.g., "@my_db.my_schema.my_stage/model.yaml")	None
semantic_model	str | None	Full semantic model YAML as string	None
semantic_view	str | None	Fully qualified semantic view name (e.g., "MY_DB.MY_SCHEMA.MY_VIEW")	None
semantic_models	list[dict[str, Any]] | None	List of semantic model/view dicts for multi-model selection (e.g., [{"semantic_view": "DB.SCH.VIEW1"}, {"semantic_model_file": "@stage/model.yaml"}])	None
messages	list[dict[str, Any]] | None	Full messages array for multi-turn conversations (if not provided, will be built from question)	None

Returns:

Name	Type	Description
AnalystResponse	AnalystResponse	Response object with SQL, text, and streaming support

Examples:

# With semantic model file
response = analyst.message(
    "Which company had most revenue?",
    semantic_model_file="@my_stage/model.yaml"
)

# Multi-turn conversation
response1 = analyst.message(
    "What's total revenue?",
    semantic_model_file="@stage/model.yaml"
)
response2 = analyst.message(
    "How does that compare to last year?",
    semantic_model_file="@stage/model.yaml",
    messages=response1.conversation_messages
)

Source code in cortex_agents/analyst.py

def message(
    self,
    question: str,
    semantic_model_file: str | None = None,
    semantic_model: str | None = None,
    semantic_view: str | None = None,
    semantic_models: list[dict[str, Any]] | None = None,
    messages: list[dict[str, Any]] | None = None,
    stream: bool = True,
) -> AnalystResponse:
    """
    Send a message to Cortex Analyst to generate SQL from natural language.

    You must specify ONE of: semantic_model_file, semantic_model, semantic_view, or semantic_models.

    Args:
        question: The natural language question (str)
        semantic_model_file: Path to semantic model YAML file on stage
                             (e.g., "@my_db.my_schema.my_stage/model.yaml")
        semantic_model: Full semantic model YAML as string
        semantic_view: Fully qualified semantic view name
                      (e.g., "MY_DB.MY_SCHEMA.MY_VIEW")
        semantic_models: List of semantic model/view dicts for multi-model selection
                        (e.g., [{"semantic_view": "DB.SCH.VIEW1"}, {"semantic_model_file": "@stage/model.yaml"}])
        messages: Full messages array for multi-turn conversations
                 (if not provided, will be built from question)

    Returns:
        AnalystResponse: Response object with SQL, text, and streaming support

    Examples:
        ```python
        # With semantic model file
        response = analyst.message(
            "Which company had most revenue?",
            semantic_model_file="@my_stage/model.yaml"
        )

        # Multi-turn conversation
        response1 = analyst.message(
            "What's total revenue?",
            semantic_model_file="@stage/model.yaml"
        )
        response2 = analyst.message(
            "How does that compare to last year?",
            semantic_model_file="@stage/model.yaml",
            messages=response1.conversation_messages
        )
        ```
    """
    if not isinstance(question, str):
        raise SnowflakeAPIError("question must be a string")

    if messages is None:
        default_messages = [{"role": "user", "content": [{"type": "text", "text": question}]}]
        try:
            payload_messages = normalize_analyst_messages(default_messages)
        except ValueError as exc:
            raise SnowflakeAPIError(str(exc)) from exc
    else:
        messages_with_question = deepcopy(messages)
        messages_with_question.append({"role": "user", "content": [{"type": "text", "text": question}]})
        try:
            payload_messages = normalize_analyst_messages(messages_with_question)
        except ValueError as exc:
            raise SnowflakeAPIError(str(exc)) from exc

    payload: dict[str, Any] = {"messages": payload_messages}

    model_flags = [bool(semantic_model_file), bool(semantic_model), bool(semantic_view), bool(semantic_models)]
    if sum(model_flags) != 1:
        raise SnowflakeAPIError(
            "You must provide exactly one of semantic_model_file, semantic_model, semantic_view, or semantic_models"
        )

    if semantic_model_file:
        payload["semantic_model_file"] = semantic_model_file
    elif semantic_model:
        payload["semantic_model"] = semantic_model
    elif semantic_view:
        payload["semantic_view"] = semantic_view
    elif semantic_models:
        payload["semantic_models"] = semantic_models

    if stream:
        payload["stream"] = True

    response_payload = self._transport.post("cortex/analyst/message", payload, stream=stream)
    return AnalystResponse(response_payload, stream=stream, request_messages=payload_messages)

submit_feedback

submit_feedback(request_id: str, positive: bool, feedback_message: str | None = None) -> dict[str, Any]

Submit feedback about an Analyst response.

Parameters:

Name	Type	Description	Default
request_id	str	The request ID from response.request_id.	required
positive	bool	Whether the feedback is positive (True) or negative (False).	required
feedback_message	str | None	Optional detailed feedback message.	None

Returns:

Type	Description
dict[str, Any]	Confirmation response (typically empty dict on success).

Examples:

analyst.submit_feedback(
    request_id=response.request_id,
    positive=True,
    feedback_message="Perfect SQL generation!"
)

Source code in cortex_agents/analyst.py

def submit_feedback(self, request_id: str, positive: bool, feedback_message: str | None = None) -> dict[str, Any]:
    """Submit feedback about an Analyst response.

    Args:
        request_id: The request ID from response.request_id.
        positive: Whether the feedback is positive (True) or negative (False).
        feedback_message: Optional detailed feedback message.

    Returns:
        Confirmation response (typically empty dict on success).

    Examples:
        ```python
        analyst.submit_feedback(
            request_id=response.request_id,
            positive=True,
            feedback_message="Perfect SQL generation!"
        )
        ```
    """
    payload: dict[str, Any] = {"request_id": request_id, "positive": positive}
    if feedback_message:
        payload["feedback_message"] = feedback_message
    return self._transport.post("cortex/analyst/feedback", payload)

suggest_questions

suggest_questions(semantic_model_file: str | None = None, semantic_view: str | None = None, max_questions: int = 3) -> list[str]

Get suggested questions based on semantic model.

Parameters:

Name	Type	Description	Default
semantic_model_file	str | None	Path to semantic model file	None
semantic_view	str | None	Name of semantic view	None
max_questions	int	Maximum number of questions to return	3

Returns:

Type	Description
list[str]	list[str]: Suggested questions

Source code in cortex_agents/analyst.py

def suggest_questions(
    self,
    semantic_model_file: str | None = None,
    semantic_view: str | None = None,
    max_questions: int = 3,
) -> list[str]:
    """Get suggested questions based on semantic model.

    Args:
        semantic_model_file: Path to semantic model file
        semantic_view: Name of semantic view
        max_questions: Maximum number of questions to return

    Returns:
        list[str]: Suggested questions
    """
    if not semantic_model_file and not semantic_view:
        raise SnowflakeAPIError("Must provide either semantic_model_file or semantic_view")

    payload: dict[str, Any] = {"max_questions": max_questions}
    if semantic_model_file:
        payload["semantic_model_file"] = semantic_model_file
    elif semantic_view:
        payload["semantic_view"] = semantic_view

    result = self._transport.post("cortex/analyst/suggest-questions", payload)
    return result.get("questions", [])

validate_semantic_model

validate_semantic_model(semantic_model_file: str | None = None, semantic_view: str | None = None) -> dict[str, Any]

Validate a semantic model.

Parameters:

Name	Type	Description	Default
semantic_model_file	str | None	Path to semantic model file	None
semantic_view	str | None	Name of semantic view	None

Returns:

Name	Type	Description
Dict	dict[str, Any]	Validation results

Source code in cortex_agents/analyst.py

def validate_semantic_model(
    self, semantic_model_file: str | None = None, semantic_view: str | None = None
) -> dict[str, Any]:
    """Validate a semantic model.

    Args:
        semantic_model_file: Path to semantic model file
        semantic_view: Name of semantic view

    Returns:
        Dict: Validation results
    """
    if not semantic_model_file and not semantic_view:
        raise SnowflakeAPIError("Must provide either semantic_model_file or semantic_view")

    payload: dict[str, Any] = {}
    if semantic_model_file:
        payload["semantic_model_file"] = semantic_model_file
    elif semantic_view:
        payload["semantic_view"] = semantic_view

    return self._transport.post("cortex/analyst/validate", payload)