Core Run Management

cortex_agents.core.run.AgentRun

AgentRun(transport: SyncTransport)

Execute synchronous Cortex Agent runs.

Source code in cortex_agents/core/run.py

def __init__(self, transport: SyncTransport) -> None:
    self._transport = transport

run

run(*, query: str | None = None, agent_name: str | None = None, database: str | None = None, schema: str | None = None, thread_id: str | int | None = None, parent_message_id: str | int | None = None, tool_choice: dict[str, Any] | None = None, messages: Iterable[dict[str, Any]] | None = None, agent_config: Mapping[str, Any] | None = None, stream: bool = True) -> AgentResponse

Execute a Cortex Agent run.

Parameters:

Name	Type	Description	Default
query	str | None	User query string.	None
agent_name	str | None	Name of saved agent to run (optional).	None
database	str | None	Database name (required if agent_name provided).	None
schema	str | None	Schema name (required if agent_name provided).	None
thread_id	str | int | None	Thread ID for multi-turn conversations (optional).	None
parent_message_id	str | int | None	Parent message ID in thread (required if thread_id set).	None
tool_choice	dict[str, Any] | None	Tool selection strategy (optional).	None
messages	Iterable[dict[str, Any]] | None	Conversation history messages (optional).	None
agent_config	Mapping[str, Any] | None	Inline agent configuration for ephemeral runs (optional).	None

Returns:

Type	Description
AgentResponse	AgentResponse with streaming support.

Raises:

Type	Description
SnowflakeAPIError	If the request fails or parameters are invalid.

Examples:

# Run saved agent
response = runner.run(
    query="What's the revenue?",
    agent_name="MY_AGENT",
    database="DB",
    schema="SCHEMA"
)

# Stream events
for event in response:
    print(event)

Source code in cortex_agents/core/run.py

def run(
    self,
    *,
    query: str | None = None,
    agent_name: str | None = None,
    database: str | None = None,
    schema: str | None = None,
    thread_id: str | int | None = None,
    parent_message_id: str | int | None = None,
    tool_choice: dict[str, Any] | None = None,
    messages: Iterable[dict[str, Any]] | None = None,
    agent_config: Mapping[str, Any] | None = None,
    stream: bool = True,
) -> AgentResponse:
    """Execute a Cortex Agent run.

    Args:
        query: User query string.
        agent_name: Name of saved agent to run (optional).
        database: Database name (required if agent_name provided).
        schema: Schema name (required if agent_name provided).
        thread_id: Thread ID for multi-turn conversations (optional).
        parent_message_id: Parent message ID in thread (required if thread_id set).
        tool_choice: Tool selection strategy (optional).
        messages: Conversation history messages (optional).
        agent_config: Inline agent configuration for ephemeral runs (optional).

    Returns:
        AgentResponse with streaming support.

    Raises:
        SnowflakeAPIError: If the request fails or parameters are invalid.

    Examples:
        ```python
        # Run saved agent
        response = runner.run(
            query="What's the revenue?",
            agent_name="MY_AGENT",
            database="DB",
            schema="SCHEMA"
        )

        # Stream events
        for event in response:
            print(event)
        ```
    """
    request = build_run_request(
        query=query,
        agent_name=agent_name,
        database=database,
        schema=schema,
        tool_choice=tool_choice,
        messages=messages,
        thread_id=thread_id,
        parent_message_id=parent_message_id,
        inline_config=dict(agent_config) if agent_config else None,
        stream=stream,
    )

    transport_result = self._transport.post(request.endpoint, request.payload, stream=stream)
    return AgentResponse(transport_result, stream=stream)

cortex_agents.core.run.AsyncAgentRun

AsyncAgentRun(transport: AsyncTransport)

Execute asynchronous Cortex Agent runs.

Source code in cortex_agents/core/run.py

def __init__(self, transport: AsyncTransport) -> None:
    self._transport = transport

run async

run(*, query: str | None = None, agent_name: str | None = None, database: str | None = None, schema: str | None = None, thread_id: str | int | None = None, parent_message_id: str | int | None = None, tool_choice: dict[str, Any] | None = None, messages: Iterable[dict[str, Any]] | None = None, agent_config: Mapping[str, Any] | None = None, stream: bool = True) -> AgentResponse

Execute a Cortex Agent run (async).

Parameters:

Name	Type	Description	Default
query	str | None	User query string.	None
agent_name	str | None	Name of saved agent to run (optional).	None
database	str | None	Database name (required if agent_name provided).	None
schema	str | None	Schema name (required if agent_name provided).	None
thread_id	str | int | None	Thread ID for multi-turn conversations (optional).	None
parent_message_id	str | int | None	Parent message ID in thread (required if thread_id set).	None
tool_choice	dict[str, Any] | None	Tool selection strategy (optional).	None
messages	Iterable[dict[str, Any]] | None	Conversation history messages (optional).	None
agent_config	Mapping[str, Any] | None	Inline agent configuration for ephemeral runs (optional).	None

Returns:

Type	Description
AgentResponse	AgentResponse with async streaming support.

Raises:

Type	Description
SnowflakeAPIError	If the request fails or parameters are invalid.

Examples:

# Run saved agent
response = await runner.run(
    query="What's the revenue?",
    agent_name="MY_AGENT",
    database="DB",
    schema="SCHEMA"
)

# Stream events asynchronously
async for event in response:
    print(event)

Source code in cortex_agents/core/run.py

async def run(
    self,
    *,
    query: str | None = None,
    agent_name: str | None = None,
    database: str | None = None,
    schema: str | None = None,
    thread_id: str | int | None = None,
    parent_message_id: str | int | None = None,
    tool_choice: dict[str, Any] | None = None,
    messages: Iterable[dict[str, Any]] | None = None,
    agent_config: Mapping[str, Any] | None = None,
    stream: bool = True,
) -> AgentResponse:
    """Execute a Cortex Agent run (async).

    Args:
        query: User query string.
        agent_name: Name of saved agent to run (optional).
        database: Database name (required if agent_name provided).
        schema: Schema name (required if agent_name provided).
        thread_id: Thread ID for multi-turn conversations (optional).
        parent_message_id: Parent message ID in thread (required if thread_id set).
        tool_choice: Tool selection strategy (optional).
        messages: Conversation history messages (optional).
        agent_config: Inline agent configuration for ephemeral runs (optional).

    Returns:
        AgentResponse with async streaming support.

    Raises:
        SnowflakeAPIError: If the request fails or parameters are invalid.

    Examples:
        ```python
        # Run saved agent
        response = await runner.run(
            query="What's the revenue?",
            agent_name="MY_AGENT",
            database="DB",
            schema="SCHEMA"
        )

        # Stream events asynchronously
        async for event in response:
            print(event)
        ```
    """
    request = build_run_request(
        query=query,
        agent_name=agent_name,
        database=database,
        schema=schema,
        tool_choice=tool_choice,
        messages=messages,
        thread_id=thread_id,
        parent_message_id=parent_message_id,
        inline_config=dict(agent_config) if agent_config else None,
        stream=stream,
    )

    transport_result = await self._transport.post(request.endpoint, request.payload, stream=stream)
    return AgentResponse(transport_result, stream=stream, is_async=True)

cortex_agents.core.run.RunRequest dataclass

RunRequest(endpoint: str, payload: dict[str, Any])

Resolved endpoint and payload for a Cortex Agent :run invocation.

Attributes:

Name	Type	Description
endpoint	str	API endpoint URL for the run request.
payload	dict[str, Any]	Request payload containing messages and configuration.