CortexAgentManager

relationalai.agent.cortex.cortex_agent_manager

CortexAgentManager(session: snowpark.Session, config: DeploymentConfig)

Deploy and manage Cortex AI agents powered by RAI semantic models.

Handles the full lifecycle of a Snowflake Cortex agent that powers a Snowflake Intelligence (SI) experience. Non-technical SI users interact with the agent through the SI UI; this class is for the deployer/admin who creates and maintains the agent behind it.

Parameters

• session

  (snowflake.snowpark.Session) - Authenticated snowpark.Session with access to the target database and schema.

• config

  (DeploymentConfig) - Configuration specifying where and how to deploy the agent.

Examples

>>> session = create_config().get_session(SnowflakeConnection)
>>> manager = CortexAgentManager(
...     session=session,
...     config=DeploymentConfig(
...         agent_name="EXAMPLE_CORTEX",
...         database="EXAMPLE",
...         schema="CORTEX",
...         warehouse="DEMOWAREHOUSE",
...     ),
... )
>>> def init_tools(model):
...     init_model(model)
...     return ToolRegistry().add(
...         model=model, description="Customers and orders")
>>> manager.deploy(
...     init_tools=init_tools, imports=discover_imports())

Methods

.deploy()

CortexAgentManager.deploy(
    init_tools: Callable,
    imports: List[Tuple[str, str]],
    base_agent_spec: Optional[CortexAgentApi.AgentSpec] = None,
    extra_packages: Optional[List[str]] = None,
) -> CortexAgentApi.AgentSpec

Deploy the agent and all required Snowflake resources.

Creates a stage (if manage_stage=True), registers RAI tool stored procedures, and deploys the Cortex agent. On failure, attempts cleanup of partially deployed resources before raising.

Parameters:

• init_tools

  (Callable) - Callable accepting a Model and returning a ToolRegistry. The framework creates a Model (with the sproc’s session) and passes it to this callable. The callable should initialize concepts/properties on the model and return a configured ToolRegistry.

• imports

  (list of tuple of (str, str)) - Project code to upload to Snowflake. Use discover_imports() for automatic recursive discovery. Format: list of (source_path, module_name) tuples.

• base_agent_spec

  (relationalai.agent.cortex.api.agent.CortexAgentApi.AgentSpec, default: None) - Base agent spec to extend with RAI tools. Default: None (creates a new spec).

• extra_packages

  (list of str, default: None) - Additional PyPI packages for the sproc environment. relationalai is included automatically.

Returns:

• relationalai.agent.cortex.api.agent.CortexAgentApi.AgentSpec - The deployed agent specification.

Raises:

• RAIException - If deployment fails at any stage.

.update()

CortexAgentManager.update(
    init_tools: Callable,
    imports: List[Tuple[str, str]],
    extra_packages: Optional[List[str]] = None,
) -> CortexAgentApi.AgentSpec

Update the agent’s tools without recreating the agent.

Recreates stored procedures with updated tool definitions and updates the agent configuration in Cortex. The agent name and identity are preserved; existing SI conversations are unaffected.

Parameters:

• init_tools

  (Callable) - Updated callable accepting a Model and returning a ToolRegistry. Same contract as deploy().

• imports

  (list of tuple of (str, str)) - Updated project code. Use discover_imports().

• extra_packages

  (list of str, default: None) - Updated PyPI package list.

Returns:

• relationalai.agent.cortex.api.agent.CortexAgentApi.AgentSpec - Updated agent specification.

Raises:

• RAIException - If agent is not deployed (call deploy() first).

.chat()

CortexAgentManager.chat() -> CortexAgentChat

Get a programmatic chat interface for the deployed agent.

Returns a CortexAgentChat that maintains conversation state across multiple send() calls. This is primarily useful for testing and automation — SI users interact through the Snowflake Intelligence UI.

Returns:

• CortexAgentChat - Chat interface instance.

Raises:

• RAIException - If agent is not deployed (call deploy() first).

.status()

CortexAgentManager.status() -> DeploymentStatus

Check deployment status of all agent components.

Queries Snowflake and Cortex to verify existence of the agent, stage, and all stored procedures.

Returns:

• DeploymentStatus - Use fully_deployed(), partially_deployed(), or clean() for quick checks.

.cleanup()

CortexAgentManager.cleanup(silent: bool = True) -> None

Remove all agent resources from Snowflake and Cortex.

Deletes the Cortex agent, all RAI tool stored procedures, and the stage (if manage_stage=True). SI conversation history is permanently lost.

Parameters:

• silent

  (bool, default: True) - Ignore errors for non-existent resources. Default: True.

Returns:

• None