Implementation:CrewAIInc CrewAI Snowflake Search Tool

Knowledge Sources	CrewAI
Domains	Tools, Database, Data_Warehouse
Last Updated	2026-02-11 00:00 GMT

Overview

SnowflakeSearchTool executes SQL queries against Snowflake data warehouses with async connection pooling, retry logic, caching, and support for both password and key-pair authentication.

Description

The SnowflakeSearchTool extends BaseTool and provides enterprise-grade Snowflake integration. It defines a SnowflakeConfig Pydantic model for connection credentials (account, user, password as SecretStr, optional private_key_path, warehouse, database, schema, role, and session parameters) with validation ensuring either password or private_key_path is provided. The tool manages an async connection pool protected by asyncio.Lock with a ThreadPoolExecutor for blocking connection creation. It supports key-pair authentication by loading PEM private keys via the cryptography library. Query execution in _execute_query implements retry logic with exponential backoff for DatabaseError and OperationalError, and optional result caching via a module-level dictionary keyed by account, database, schema, query, and timeout. The async _run method supports database and schema overrides before query execution. Cleanup in __del__ closes pooled connections and shuts down the thread pool.

Usage

Use this tool when a CrewAI agent needs to query Snowflake data warehouses for data analysis, business intelligence, or data retrieval workflows requiring SQL execution against enterprise data sources.

Code Reference

Source Location

• Repository: CrewAI
• File: lib/crewai-tools/src/crewai_tools/tools/snowflake_search_tool/snowflake_search_tool.py
• Lines: 1-289

Signature

class SnowflakeConfig(BaseModel):
    account: str = Field(..., description="Snowflake account identifier")
    user: str = Field(..., description="Snowflake username")
    password: SecretStr | None = Field(None)
    private_key_path: str | None = Field(None)
    warehouse: str | None = Field(None)
    database: str | None = Field(None)
    snowflake_schema: str | None = Field(None)
    role: str | None = Field(None)

class SnowflakeSearchToolInput(BaseModel):
    query: str = Field(..., description="SQL query or semantic search query to execute")
    database: str | None = Field(None)
    snowflake_schema: str | None = Field(None)
    timeout: int | None = Field(300)

class SnowflakeSearchTool(BaseTool):
    name: str = "Snowflake Database Search"
    description: str = "Execute SQL queries or semantic search on Snowflake data warehouse. ..."
    args_schema: type[BaseModel] = SnowflakeSearchToolInput
    config: SnowflakeConfig = Field(...)
    pool_size: int = Field(default=5)
    max_retries: int = Field(default=3)
    retry_delay: float = Field(default=1.0)
    enable_caching: bool = Field(default=True)

    async def _run(self, query: str, database: str | None = None,
                   snowflake_schema: str | None = None, timeout: int = 300, **kwargs) -> Any:
        ...

Import

from crewai_tools import SnowflakeSearchTool

I/O Contract

Inputs

Name	Type	Required	Description
query	str	Yes	SQL query or semantic search query to execute
database	str	No	Override default database for this query
snowflake_schema	str	No	Override default schema for this query
timeout	int	No	Query timeout in seconds (default 300)

Outputs

Name	Type	Description
_run() returns	list[dict[str, Any]]	List of dictionaries representing query result rows with column names as keys

Usage Examples

Basic Usage

from crewai_tools import SnowflakeSearchTool

config = {
    "account": "my_account",
    "user": "my_user",
    "password": "my_password",
    "warehouse": "my_warehouse",
    "database": "my_database",
    "snowflake_schema": "public",
}
tool = SnowflakeSearchTool(config=config)

Related Pages

• Principle:CrewAIInc_CrewAI_Built_In_Tool_Selection