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 returning a ToolRegistry. Two forms are supported:

  Recommended (0-param): Import your model definition code inside the function body. The import must happen inside init_tools so the Model is created within the sproc’s active Snowpark session:

  def init_tools():
      from myproject.model import core
      return ToolRegistry().add(
          model=core.model, description='...')

  Legacy (1-param): Accept a Model from the framework:

  def init_tools(model):
      return ToolRegistry().add(
          model=model, description='...')

• 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 returning a ToolRegistry. Same contract as deploy() — supports both 0-param (recommended) and 1-param (legacy) forms.

• 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