Core Entity Management

cortex_agents.core.entity.AgentEntity

AgentEntity(transport: Any)

Manage Cortex Agent entities synchronously.

Source code in cortex_agents/core/entity.py

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

create_agent

create_agent(name: str, config: dict[str, Any], database: str, schema: str, *, create_mode: str | None = None) -> dict[str, Any]

Create a new Cortex Agent in Snowflake.

Parameters:

Name	Type	Description	Default
name	str	Agent name.	required
config	dict[str, Any]	Agent configuration including instructions, models, tools, etc.	required
database	str	Database name where the agent will be created.	required
schema	str	Schema name where the agent will be created.	required
create_mode	str | None	Optional creation mode (e.g., 'or_replace').	None

Returns:

Type	Description
dict[str, Any]	Response dictionary with creation status.

Raises:

Type	Description
SnowflakeAPIError	If the API request fails.

Source code in cortex_agents/core/entity.py

def create_agent(
    self,
    name: str,
    config: dict[str, Any],
    database: str,
    schema: str,
    *,
    create_mode: str | None = None,
) -> dict[str, Any]:
    """Create a new Cortex Agent in Snowflake.

    Args:
        name: Agent name.
        config: Agent configuration including instructions, models, tools, etc.
        database: Database name where the agent will be created.
        schema: Schema name where the agent will be created.
        create_mode: Optional creation mode (e.g., 'or_replace').

    Returns:
        Response dictionary with creation status.

    Raises:
        SnowflakeAPIError: If the API request fails.
    """
    endpoint = f"databases/{database}/schemas/{schema}/agents"
    params = {"createMode": create_mode} if create_mode else None

    payload: dict[str, Any] = {"name": name, **config}
    return self._transport.post(endpoint, payload, params=params)

get_agent

get_agent(name: str, database: str, schema: str) -> dict[str, Any]

Retrieve details of an existing Cortex Agent.

Parameters:

Name	Type	Description	Default
name	str	Agent name.	required
database	str	Database name.	required
schema	str	Schema name.	required

Returns:

Type	Description
dict[str, Any]	Agent configuration and metadata.

Raises:

Type	Description
SnowflakeAPIError	If the agent doesn't exist or request fails.

Source code in cortex_agents/core/entity.py

def get_agent(self, name: str, database: str, schema: str) -> dict[str, Any]:
    """Retrieve details of an existing Cortex Agent.

    Args:
        name: Agent name.
        database: Database name.
        schema: Schema name.

    Returns:
        Agent configuration and metadata.

    Raises:
        SnowflakeAPIError: If the agent doesn't exist or request fails.
    """
    endpoint = f"databases/{database}/schemas/{schema}/agents/{name}"
    return self._transport.get(endpoint)

update_agent

update_agent(name: str, config: dict[str, Any], database: str, schema: str) -> dict[str, Any]

Update an existing Cortex Agent's configuration.

Parameters:

Name	Type	Description	Default
name	str	Agent name.	required
config	dict[str, Any]	Updated agent configuration.	required
database	str	Database name.	required
schema	str	Schema name.	required

Returns:

Type	Description
dict[str, Any]	Response dictionary with update status.

Raises:

Type	Description
SnowflakeAPIError	If the agent doesn't exist or request fails.

Source code in cortex_agents/core/entity.py

def update_agent(self, name: str, config: dict[str, Any], database: str, schema: str) -> dict[str, Any]:
    """Update an existing Cortex Agent's configuration.

    Args:
        name: Agent name.
        config: Updated agent configuration.
        database: Database name.
        schema: Schema name.

    Returns:
        Response dictionary with update status.

    Raises:
        SnowflakeAPIError: If the agent doesn't exist or request fails.
    """
    endpoint = f"databases/{database}/schemas/{schema}/agents/{name}"
    return self._transport.put(endpoint, config)

list_agents

list_agents(database: str, schema: str, *, like: str | None = None, from_name: str | None = None, limit: int | None = None) -> list[dict[str, Any]]

List all Cortex Agents in a schema.

Parameters:

Name	Type	Description	Default
database	str	Database name.	required
schema	str	Schema name.	required
like	str | None	SQL LIKE pattern to filter agent names (optional).	None
from_name	str | None	Starting agent name for pagination (optional).	None
limit	int | None	Maximum number of agents to return (optional).	None

Returns:

Type	Description
list[dict[str, Any]]	List of agent metadata dictionaries.

Raises:

Type	Description
SnowflakeAPIError	If the request fails.

Source code in cortex_agents/core/entity.py

def list_agents(
    self,
    database: str,
    schema: str,
    *,
    like: str | None = None,
    from_name: str | None = None,
    limit: int | None = None,
) -> list[dict[str, Any]]:
    """List all Cortex Agents in a schema.

    Args:
        database: Database name.
        schema: Schema name.
        like: SQL LIKE pattern to filter agent names (optional).
        from_name: Starting agent name for pagination (optional).
        limit: Maximum number of agents to return (optional).

    Returns:
        List of agent metadata dictionaries.

    Raises:
        SnowflakeAPIError: If the request fails.
    """
    endpoint = f"databases/{database}/schemas/{schema}/agents"
    params: dict[str, Any] = {}
    if like:
        params["like"] = like
    if from_name:
        params["fromName"] = from_name
    if limit:
        params["showLimit"] = limit

    return self._transport.get(endpoint, params if params else None)

delete_agent

delete_agent(name: str, database: str, schema: str, *, if_exists: bool = False) -> dict[str, Any]

Delete a Cortex Agent.

Parameters:

Name	Type	Description	Default
name	str	Agent name.	required
database	str	Database name.	required
schema	str	Schema name.	required
if_exists	bool	If True, don't raise error if agent doesn't exist.	False

Returns:

Type	Description
dict[str, Any]	Response dictionary with deletion status.

Raises:

Type	Description
SnowflakeAPIError	If the agent doesn't exist (when if_exists=False) or request fails.

Source code in cortex_agents/core/entity.py

def delete_agent(
    self,
    name: str,
    database: str,
    schema: str,
    *,
    if_exists: bool = False,
) -> dict[str, Any]:
    """Delete a Cortex Agent.

    Args:
        name: Agent name.
        database: Database name.
        schema: Schema name.
        if_exists: If True, don't raise error if agent doesn't exist.

    Returns:
        Response dictionary with deletion status.

    Raises:
        SnowflakeAPIError: If the agent doesn't exist (when if_exists=False)
            or request fails.
    """
    endpoint = f"databases/{database}/schemas/{schema}/agents/{name}"
    params = {"ifExists": "true"} if if_exists else None
    return self._transport.delete(endpoint, params)

cortex_agents.core.entity.AsyncAgentEntity

AsyncAgentEntity(transport: Any)

Manage Cortex Agent entities asynchronously.

Source code in cortex_agents/core/entity.py

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

create_agent async

create_agent(name: str, config: dict[str, Any], database: str, schema: str, *, create_mode: str | None = None) -> dict[str, Any]

Create a new Cortex Agent in Snowflake (async).

Parameters:

Name	Type	Description	Default
name	str	Agent name.	required
config	dict[str, Any]	Agent configuration including instructions, models, tools, etc.	required
database	str	Database name where the agent will be created.	required
schema	str	Schema name where the agent will be created.	required
create_mode	str | None	Optional creation mode (e.g., 'or_replace').	None

Returns:

Type	Description
dict[str, Any]	Response dictionary with creation status.

Raises:

Type	Description
SnowflakeAPIError	If the API request fails.

Source code in cortex_agents/core/entity.py

async def create_agent(
    self,
    name: str,
    config: dict[str, Any],
    database: str,
    schema: str,
    *,
    create_mode: str | None = None,
) -> dict[str, Any]:
    """Create a new Cortex Agent in Snowflake (async).

    Args:
        name: Agent name.
        config: Agent configuration including instructions, models, tools, etc.
        database: Database name where the agent will be created.
        schema: Schema name where the agent will be created.
        create_mode: Optional creation mode (e.g., 'or_replace').

    Returns:
        Response dictionary with creation status.

    Raises:
        SnowflakeAPIError: If the API request fails.
    """
    endpoint = f"databases/{database}/schemas/{schema}/agents"
    params = {"createMode": create_mode} if create_mode else None

    payload: dict[str, Any] = {"name": name, **config}
    return await self._transport.post(endpoint, payload, params=params)

get_agent async

get_agent(name: str, database: str, schema: str) -> dict[str, Any]

Retrieve details of an existing Cortex Agent (async).

Parameters:

Name	Type	Description	Default
name	str	Agent name.	required
database	str	Database name.	required
schema	str	Schema name.	required

Returns:

Type	Description
dict[str, Any]	Agent configuration and metadata.

Raises:

Type	Description
SnowflakeAPIError	If the agent doesn't exist or request fails.

Source code in cortex_agents/core/entity.py

async def get_agent(self, name: str, database: str, schema: str) -> dict[str, Any]:
    """Retrieve details of an existing Cortex Agent (async).

    Args:
        name: Agent name.
        database: Database name.
        schema: Schema name.

    Returns:
        Agent configuration and metadata.

    Raises:
        SnowflakeAPIError: If the agent doesn't exist or request fails.
    """
    endpoint = f"databases/{database}/schemas/{schema}/agents/{name}"
    return await self._transport.get(endpoint)

update_agent async

update_agent(name: str, config: dict[str, Any], database: str, schema: str) -> dict[str, Any]

Update an existing Cortex Agent's configuration (async).

Parameters:

Name	Type	Description	Default
name	str	Agent name.	required
config	dict[str, Any]	Updated agent configuration.	required
database	str	Database name.	required
schema	str	Schema name.	required

Returns:

Type	Description
dict[str, Any]	Response dictionary with update status.

Raises:

Type	Description
SnowflakeAPIError	If the agent doesn't exist or request fails.

Source code in cortex_agents/core/entity.py

async def update_agent(
    self,
    name: str,
    config: dict[str, Any],
    database: str,
    schema: str,
) -> dict[str, Any]:
    """Update an existing Cortex Agent's configuration (async).

    Args:
        name: Agent name.
        config: Updated agent configuration.
        database: Database name.
        schema: Schema name.

    Returns:
        Response dictionary with update status.

    Raises:
        SnowflakeAPIError: If the agent doesn't exist or request fails.
    """
    endpoint = f"databases/{database}/schemas/{schema}/agents/{name}"
    return await self._transport.put(endpoint, config)

list_agents async

list_agents(database: str, schema: str, *, like: str | None = None, from_name: str | None = None, limit: int | None = None) -> list[dict[str, Any]]

List all Cortex Agents in a schema (async).

Parameters:

Name	Type	Description	Default
database	str	Database name.	required
schema	str	Schema name.	required
like	str | None	SQL LIKE pattern to filter agent names (optional).	None
from_name	str | None	Starting agent name for pagination (optional).	None
limit	int | None	Maximum number of agents to return (optional).	None

Returns:

Type	Description
list[dict[str, Any]]	List of agent metadata dictionaries.

Raises:

Type	Description
SnowflakeAPIError	If the request fails.

Source code in cortex_agents/core/entity.py

async def list_agents(
    self,
    database: str,
    schema: str,
    *,
    like: str | None = None,
    from_name: str | None = None,
    limit: int | None = None,
) -> list[dict[str, Any]]:
    """List all Cortex Agents in a schema (async).

    Args:
        database: Database name.
        schema: Schema name.
        like: SQL LIKE pattern to filter agent names (optional).
        from_name: Starting agent name for pagination (optional).
        limit: Maximum number of agents to return (optional).

    Returns:
        List of agent metadata dictionaries.

    Raises:
        SnowflakeAPIError: If the request fails.
    """
    endpoint = f"databases/{database}/schemas/{schema}/agents"
    params: dict[str, Any] = {}
    if like:
        params["like"] = like
    if from_name:
        params["fromName"] = from_name
    if limit:
        params["showLimit"] = limit

    return await self._transport.get(endpoint, params if params else None)

delete_agent async

delete_agent(name: str, database: str, schema: str, *, if_exists: bool = False) -> dict[str, Any]

Delete a Cortex Agent (async).

Parameters:

Name	Type	Description	Default
name	str	Agent name.	required
database	str	Database name.	required
schema	str	Schema name.	required
if_exists	bool	If True, don't raise error if agent doesn't exist.	False

Returns:

Type	Description
dict[str, Any]	Response dictionary with deletion status.

Raises:

Type	Description
SnowflakeAPIError	If the agent doesn't exist (when if_exists=False) or request fails.

Source code in cortex_agents/core/entity.py

async def delete_agent(
    self,
    name: str,
    database: str,
    schema: str,
    *,
    if_exists: bool = False,
) -> dict[str, Any]:
    """Delete a Cortex Agent (async).

    Args:
        name: Agent name.
        database: Database name.
        schema: Schema name.
        if_exists: If True, don't raise error if agent doesn't exist.

    Returns:
        Response dictionary with deletion status.

    Raises:
        SnowflakeAPIError: If the agent doesn't exist (when if_exists=False)
            or request fails.
    """
    endpoint = f"databases/{database}/schemas/{schema}/agents/{name}"
    params = {"ifExists": "true"} if if_exists else None
    return await self._transport.delete(endpoint, params)