From 3eff0955d5e5c0aead9e5cfde5b982a92187dbe1 Mon Sep 17 00:00:00 2001
From: Matt Harris <mharris@parallel.ai>
Date: Sat, 24 Jan 2026 01:59:17 -0500
Subject: [PATCH 01/13] feat: Snowflake integration with PyPI package support

- Add CLI support for `parallel-cli enrich deploy --system snowflake`
- Use parallel-web-tools from PyPI via Snowflake's artifact repository
- Share core enrichment logic with BigQuery/Spark integrations
- Add comprehensive error handling (fail on CREATE statement errors)
- Simplify codebase with extracted helper functions
- Update docs with MFA setup, account identifier formats, troubleshooting
---
 docs/snowflake-setup.md                       | 396 +++++++++++++-----
 parallel_web_tools/cli/commands.py            | 102 ++++-
 parallel_web_tools/core/batch.py              | 251 +++++------
 .../integrations/snowflake/__init__.py        |   2 +-
 .../integrations/snowflake/deploy.py          | 222 +++++-----
 .../integrations/snowflake/sql/01_setup.sql   |  12 +-
 .../snowflake/sql/02_create_udf.sql           | 188 +--------
 7 files changed, 636 insertions(+), 537 deletions(-)

diff --git a/docs/snowflake-setup.md b/docs/snowflake-setup.md
index 80ebc35..8952df3 100644
--- a/docs/snowflake-setup.md
+++ b/docs/snowflake-setup.md
@@ -24,9 +24,48 @@ The integration uses Snowflake's External Access Integration feature to allow UD
 
 ## Prerequisites
 
-1. **Snowflake Account** with ACCOUNTADMIN privileges (or equivalent)
-2. **Python 3.12+** (for Python deployment)
-3. **Parallel API Key** from [platform.parallel.ai](https://platform.parallel.ai)
+1. **Snowflake Account** - Paid account required (trial accounts don't support External Access)
+2. **ACCOUNTADMIN Role** - Required for creating integrations (see [Manual Deployment](#manual-sql-deployment-for-admins) if you don't have this)
+3. **MFA Setup** - If your account requires MFA, you'll need an authenticator app configured
+4. **Python 3.12+** with `parallel-web-tools[snowflake]` installed
+5. **Parallel API Key** from [platform.parallel.ai](https://platform.parallel.ai)
+
+## How It Works
+
+The Snowflake integration uses the `parallel-web-tools` package from PyPI, sharing the same core enrichment logic as BigQuery and Spark integrations:
+
+```
+Snowflake UDF
+    │
+    ├── Uses: ARTIFACT_REPOSITORY = snowflake.snowpark.pypi_shared_repository
+    ├── Package: parallel-web-tools (from PyPI)
+    │
+    └── Calls: enrich_batch() from parallel_web_tools.core
+                    │
+                    └── Parallel Task Group API
+```
+
+This ensures consistent behavior across all platforms.
+
+## Finding Your Account Identifier
+
+Your Snowflake account identifier is in your Snowsight URL:
+
+```
+https://app.snowflake.com/ORGNAME/ACCOUNTNAME/worksheets
+                         ↑       ↑
+                         └───┬───┘
+                     Account: ORGNAME-ACCOUNTNAME
+```
+
+**Examples:**
+- URL: `https://app.snowflake.com/myorg/myaccount/` → Account: `myorg-myaccount`
+- URL: `https://app.snowflake.com/us-east-1/xy12345/` → Account: `xy12345.us-east-1`
+
+You can also run this SQL in Snowsight:
+```sql
+SELECT CURRENT_ORGANIZATION_NAME() || '-' || CURRENT_ACCOUNT_NAME() AS account_identifier;
+```
 
 ## Installation
 
@@ -34,18 +73,90 @@ The integration uses Snowflake's External Access Integration feature to allow UD
 pip install parallel-web-tools[snowflake]
 ```
 
-## Quick Start - Python Deployment
+Or with uv:
+```bash
+uv add "parallel-web-tools[snowflake]"
+```
+
+## Quick Start - CLI Deployment
+
+The easiest way to deploy is using the CLI:
+
+### Basic (Password Auth)
+
+```bash
+parallel-cli enrich deploy --system snowflake \
+    --account ORGNAME-ACCOUNTNAME \
+    --user your-username \
+    --password "your-password" \
+    --warehouse COMPUTE_WH
+```
 
-The easiest way to deploy is using the Python helper:
+### With MFA
+
+If your account requires MFA, you need to:
+
+1. **Set up an authenticator app** (Google Authenticator, Duo, etc.) in Snowsight:
+   - Go to your profile → Security → Multi-factor Authentication
+   - Enroll your authenticator app
+
+2. **Run with the passcode flag:**
+```bash
+parallel-cli enrich deploy --system snowflake \
+    --account ORGNAME-ACCOUNTNAME \
+    --user your-username \
+    --password "your-password" \
+    --authenticator username_password_mfa \
+    --passcode 123456 \
+    --warehouse COMPUTE_WH
+```
+
+Replace `123456` with the current 6-digit code from your authenticator app. Run the command immediately after getting a fresh code (they expire quickly).
+
+### Environment Variables
+
+To avoid passing sensitive values on the command line:
+
+```bash
+# Use single quotes if password contains special characters like !
+export SNOWFLAKE_PASSWORD='your!password'
+export PARALLEL_API_KEY='your-api-key'
+
+parallel-cli enrich deploy --system snowflake \
+    --account ORGNAME-ACCOUNTNAME \
+    --user your-username \
+    --password "$SNOWFLAKE_PASSWORD" \
+    --authenticator username_password_mfa \
+    --passcode 123456 \
+    --warehouse COMPUTE_WH
+```
+
+### CLI Options
+
+| Option | Default | Description |
+|--------|---------|-------------|
+| `--account` | required | Snowflake account identifier |
+| `--user` | required | Snowflake username |
+| `--password` | - | Snowflake password |
+| `--warehouse` | `COMPUTE_WH` | Warehouse to use |
+| `--role` | `ACCOUNTADMIN` | Role for deployment |
+| `--authenticator` | `externalbrowser` | Auth method |
+| `--passcode` | - | MFA code from authenticator app |
+| `--api-key` | env var | Parallel API key |
+
+## Quick Start - Python Deployment
 
 ```python
 from parallel_web_tools.integrations.snowflake import deploy_parallel_functions
 
 deploy_parallel_functions(
-    account="your-account.us-east-1",
+    account="orgname-accountname",
     user="your-user",
     password="your-password",
     parallel_api_key="your-parallel-api-key",
+    # For MFA:
+    authenticator="username_password_mfa",
+    passcode="123456",
 )
 ```
 
@@ -58,39 +169,55 @@ This creates:
 - `parallel_enrich()` UDF
 - Roles: `PARALLEL_DEVELOPER` and `PARALLEL_USER`
 
-## Quick Start - Manual SQL Deployment
+## Manual SQL Deployment (For Admins)
 
-If you prefer to run SQL manually:
+If you don't have ACCOUNTADMIN access, ask your Snowflake admin to run the setup SQL.
 
-### Step 1: Get SQL Templates
+### Generate SQL for Admin
 
-```python
+```bash
+# Generate the SQL scripts
+python -c "
 from parallel_web_tools.integrations.snowflake import get_setup_sql, get_udf_sql
 
-# Get setup SQL with your API key
-setup_sql = get_setup_sql(api_key="your-parallel-api-key")
-print(setup_sql)
-
-# Get UDF creation SQL
-udf_sql = get_udf_sql()
-print(udf_sql)
+print('=== SETUP SQL (run first) ===')
+print(get_setup_sql('YOUR_PARALLEL_API_KEY'))
+print()
+print('=== UDF SQL (run second) ===')
+print(get_udf_sql())
+"
 ```
 
-### Step 2: Run in Snowflake
+Replace `YOUR_PARALLEL_API_KEY` with your actual API key, then send the output to your admin.
 
-Execute the SQL scripts in order:
+### What Admin Needs to Run
 
-1. Run `01_setup.sql` to create network infrastructure
-2. Run `02_create_udf.sql` to create the UDF
+1. **Run as ACCOUNTADMIN** - Required for External Access Integration
+2. **Execute setup SQL** - Creates network rule, secret, integration
+3. **Execute UDF SQL** - Creates the parallel_enrich() function
+
+After admin completes setup, users with `PARALLEL_USER` role can use the function.
 
 ## SQL Usage
 
+First, set your context or use fully qualified names:
+
+```sql
+-- Option 1: Set context
+USE DATABASE PARALLEL_INTEGRATION;
+USE SCHEMA ENRICHMENT;
+
+-- Option 2: Use fully qualified names
+SELECT PARALLEL_INTEGRATION.ENRICHMENT.parallel_enrich(...);
+```
+
 ### Basic Enrichment
 
 ```sql
-SELECT parallel_enrich(
-    OBJECT_CONSTRUCT('company_name', 'Google'),
-    ARRAY_CONSTRUCT('CEO name', 'Founding year')
+-- Returns JSON with enriched fields and basis/citations
+SELECT PARALLEL_INTEGRATION.ENRICHMENT.parallel_enrich(
+    OBJECT_CONSTRUCT('company_name', 'Google', 'website', 'google.com'),
+    ARRAY_CONSTRUCT('CEO name', 'Founding year', 'Brief description')
 ) AS enriched_data;
 ```
 
@@ -99,10 +226,28 @@ Result:
 {
   "ceo_name": "Sundar Pichai",
   "founding_year": "1998",
+  "brief_description": "Google is a multinational technology company...",
   "basis": [...]
 }
 ```
 
+### Parsing JSON Results
+
+```sql
+-- Extract fields from the JSON result
+SELECT
+    data:ceo_name::STRING AS ceo_name,
+    data:founding_year::STRING AS founding_year,
+    data:brief_description::STRING AS description,
+    data:basis AS basis
+FROM (
+    SELECT PARALLEL_INTEGRATION.ENRICHMENT.parallel_enrich(
+        OBJECT_CONSTRUCT('company_name', 'Google', 'website', 'google.com'),
+        ARRAY_CONSTRUCT('CEO name', 'Founding year', 'Brief description')
+    ) AS data
+);
+```
+
 ### Multiple Input Fields
 
 ```sql
@@ -181,6 +326,90 @@ CROSS JOIN LATERAL (
 ) e;
 ```
 
+## Troubleshooting
+
+### "External access is not supported for trial accounts"
+
+Snowflake trial accounts cannot make external HTTP calls. You need to upgrade to a paid Snowflake account (Standard edition or above).
+
+### "Failed to connect... SAML Identity Provider"
+
+SSO/SAML isn't configured for your account. Use password authentication instead:
+
+```bash
+parallel-cli enrich deploy --system snowflake \
+    --account your-account \
+    --user your-user \
+    --password "your-password" \
+    ...
+```
+
+### "Multi-factor authentication is required"
+
+Your account requires MFA. Set up an authenticator app in Snowsight, then use:
+
+```bash
+parallel-cli enrich deploy --system snowflake \
+    --authenticator username_password_mfa \
+    --passcode 123456 \
+    ...
+```
+
+### "Role 'ACCOUNTADMIN' is not granted to this user"
+
+You don't have ACCOUNTADMIN. Try a different role you have access to:
+
+```bash
+parallel-cli enrich deploy --system snowflake \
+    --role SYSADMIN \
+    ...
+```
+
+Or check your roles:
+```sql
+SHOW GRANTS TO USER your_username;
+```
+
+### "Insufficient privileges to operate on account"
+
+ACCOUNTADMIN is required for creating External Access Integrations. Either:
+1. Get ACCOUNTADMIN access from your Snowflake admin
+2. Have an admin run the SQL manually (see [Manual SQL Deployment](#manual-sql-deployment-for-admins))
+
+### "Integration does not exist or not authorized"
+
+The External Access Integration wasn't created. This usually means:
+- You don't have ACCOUNTADMIN privileges
+- The setup SQL didn't complete successfully
+
+Re-run with ACCOUNTADMIN role or have an admin run the setup.
+
+### "Package 'parallel-web-tools' not found" or PyPI errors
+
+The UDF uses `parallel-web-tools` from PyPI. Ensure:
+1. The `SNOWFLAKE.PYPI_REPOSITORY_USER` role is granted (setup SQL does this automatically)
+2. Your Snowflake account has PyPI repository access enabled
+
+```sql
+-- Verify PyPI access
+GRANT DATABASE ROLE SNOWFLAKE.PYPI_REPOSITORY_USER TO ROLE your_role;
+```
+
+### Authentication Pop-ups
+
+If you see repeated authentication prompts, install keyring support:
+
+```bash
+pip install "snowflake-connector-python[secure-local-storage]"
+```
+
+### Timeout Errors
+
+For complex enrichments, the 5-minute timeout may not be enough. Consider:
+- Using `lite-fast` processor for faster results
+- Processing fewer rows per query
+- Breaking large enrichments into batches
+
 ## API Reference
 
 ### `deploy_parallel_functions()`
@@ -196,6 +425,8 @@ def deploy_parallel_functions(
     role: str = "ACCOUNTADMIN",
     parallel_api_key: str | None = None,
     authenticator: str | None = None,
+    passcode: str | None = None,
+    force: bool = False,
 ) -> None
 ```
 
@@ -211,7 +442,9 @@ def deploy_parallel_functions(
 | `schema` | `str` | `"ENRICHMENT"` | Schema to create |
 | `role` | `str` | `"ACCOUNTADMIN"` | Role for deployment |
 | `parallel_api_key` | `str \| None` | `None` | API key (uses env var if not provided) |
-| `authenticator` | `str \| None` | `None` | Auth method (e.g., "externalbrowser") |
+| `authenticator` | `str \| None` | `None` | Auth method (e.g., "username_password_mfa") |
+| `passcode` | `str \| None` | `None` | MFA code from authenticator app |
+| `force` | `bool` | `False` | Skip confirmation for existing resources |
 
 ### `cleanup_parallel_functions()`
 
@@ -290,90 +523,6 @@ Grant PARALLEL_USER to users who need to run enrichments:
 GRANT ROLE PARALLEL_USER TO USER analyst_user;
 ```
 
-## Error Handling
-
-Errors are returned as JSON in the result:
-
-```sql
-SELECT parallel_enrich(
-    OBJECT_CONSTRUCT('company_name', 'NonexistentCompanyXYZ'),
-    ARRAY_CONSTRUCT('CEO name')
-):error::STRING AS error_message;
-```
-
-Common errors:
-- `"No API key provided"` - Secret not configured
-- `"Timeout waiting for enrichment"` - API took too long
-- `"API request failed: ..."` - Network or API error
-
-## Cleanup
-
-### Using Python
-
-```python
-from parallel_web_tools.integrations.snowflake import cleanup_parallel_functions
-
-cleanup_parallel_functions(
-    account="your-account",
-    user="your-user",
-    password="your-password",
-)
-```
-
-### Using SQL
-
-```python
-from parallel_web_tools.integrations.snowflake import get_cleanup_sql
-print(get_cleanup_sql())
-```
-
-Then execute the SQL in Snowflake.
-
-## Troubleshooting
-
-### "External access integration not found"
-
-Ensure the integration was created:
-
-```sql
-SHOW EXTERNAL ACCESS INTEGRATIONS LIKE 'parallel_api%';
-```
-
-If not found, re-run `01_setup.sql`.
-
-### "Network rule violation"
-
-The network rule may not be allowing traffic. Verify:
-
-```sql
-SHOW NETWORK RULES LIKE 'parallel_api%';
-```
-
-### "Secret not found"
-
-The API key secret may be missing:
-
-```sql
-SHOW SECRETS LIKE 'parallel_api%';
-```
-
-Re-run setup with correct API key.
-
-### Timeout Errors
-
-For complex enrichments, the 5-minute timeout may not be enough. Consider:
-- Using `lite-fast` processor for faster results
-- Processing fewer rows per query
-- Breaking large enrichments into batches
-
-### Permission Errors
-
-Ensure you have the required role:
-
-```sql
-USE ROLE ACCOUNTADMIN;  -- Or PARALLEL_DEVELOPER
-```
-
 ## Cost Considerations
 
 Each row enrichment makes one API call. Costs depend on:
@@ -426,6 +575,35 @@ SELECT company_name, parallel_enrich(...) AS data
 FROM companies;
 ```
 
+## Cleanup
+
+### Using CLI
+
+```bash
+# Not yet implemented - use Python or SQL
+```
+
+### Using Python
+
+```python
+from parallel_web_tools.integrations.snowflake import cleanup_parallel_functions
+
+cleanup_parallel_functions(
+    account="your-account",
+    user="your-user",
+    password="your-password",
+)
+```
+
+### Using SQL
+
+```python
+from parallel_web_tools.integrations.snowflake import get_cleanup_sql
+print(get_cleanup_sql())
+```
+
+Then execute the SQL in Snowflake.
+
 ## Next Steps
 
 - See the [demo notebook](../notebooks/snowflake_enrichment_demo.ipynb) for more examples
diff --git a/parallel_web_tools/cli/commands.py b/parallel_web_tools/cli/commands.py
index fb8e164..968c067 100644
--- a/parallel_web_tools/cli/commands.py
+++ b/parallel_web_tools/cli/commands.py
@@ -729,13 +729,48 @@ def enrich_suggest(intent: str, source_columns: str | None, output_json: bool):
 
 
 @enrich.command(name="deploy")
-@click.option("--system", type=click.Choice(["bigquery"]), required=True, help="Target system to deploy to")
+@click.option(
+    "--system", type=click.Choice(["bigquery", "snowflake"]), required=True, help="Target system to deploy to"
+)
 @click.option("--project", "-p", help="Cloud project ID (required for bigquery)")
-@click.option("--region", "-r", default="us-central1", show_default=True, help="Cloud region")
+@click.option("--region", "-r", default="us-central1", show_default=True, help="Cloud region (BigQuery)")
 @click.option("--api-key", "-k", help="Parallel API key (or use PARALLEL_API_KEY env var)")
 @click.option("--dataset", default="parallel_functions", show_default=True, help="Dataset name (BigQuery)")
-def enrich_deploy(system: str, project: str | None, region: str, api_key: str | None, dataset: str):
+@click.option("--account", help="Snowflake account identifier (e.g., abc12345.us-east-1)")
+@click.option("--user", "-u", help="Snowflake username")
+@click.option("--password", help="Snowflake password (or use SSO with --authenticator)")
+@click.option("--warehouse", "-w", default="COMPUTE_WH", show_default=True, help="Snowflake warehouse")
+@click.option("--authenticator", default="externalbrowser", show_default=True, help="Snowflake auth method")
+@click.option("--passcode", help="MFA passcode from authenticator app (use with --authenticator username_password_mfa)")
+@click.option("--role", default="ACCOUNTADMIN", show_default=True, help="Snowflake role for deployment")
+def enrich_deploy(
+    system: str,
+    project: str | None,
+    region: str,
+    api_key: str | None,
+    dataset: str,
+    account: str | None,
+    user: str | None,
+    password: str | None,
+    warehouse: str,
+    authenticator: str,
+    passcode: str | None,
+    role: str,
+):
     """Deploy Parallel enrichment to a cloud system."""
+    # Resolve API key for all systems
+    if not api_key:
+        api_key = os.environ.get("PARALLEL_API_KEY")
+    if not api_key:
+        try:
+            api_key = get_api_key()
+        except Exception:
+            pass
+    if not api_key:
+        console.print("[bold red]Error: Parallel API key required[/bold red]")
+        console.print("  Use --api-key, PARALLEL_API_KEY env var, or run 'parallel-cli login'")
+        raise click.Abort()
+
     if system == "bigquery":
         if not project:
             console.print("[bold red]Error: --project is required for BigQuery deployment.[/bold red]")
@@ -743,18 +778,6 @@ def enrich_deploy(system: str, project: str | None, region: str, api_key: str |
 
         from parallel_web_tools.integrations.bigquery import deploy_bigquery_integration
 
-        if not api_key:
-            api_key = os.environ.get("PARALLEL_API_KEY")
-        if not api_key:
-            try:
-                api_key = get_api_key()
-            except Exception:
-                pass
-        if not api_key:
-            console.print("[bold red]Error: Parallel API key required[/bold red]")
-            console.print("  Use --api-key, PARALLEL_API_KEY env var, or run 'parallel-cli login'")
-            raise click.Abort()
-
         console.print(f"[bold cyan]Deploying to BigQuery in {project}...[/bold cyan]\n")
 
         try:
@@ -772,6 +795,55 @@ def enrich_deploy(system: str, project: str | None, region: str, api_key: str |
             console.print(f"[bold red]Deployment failed: {e}[/bold red]")
             raise click.Abort() from None
 
+    elif system == "snowflake":
+        if not account:
+            console.print("[bold red]Error: --account is required for Snowflake deployment.[/bold red]")
+            raise click.Abort()
+        if not user:
+            console.print("[bold red]Error: --user is required for Snowflake deployment.[/bold red]")
+            raise click.Abort()
+
+        from parallel_web_tools.integrations.snowflake import deploy_parallel_functions
+
+        console.print(f"[bold cyan]Deploying to Snowflake account {account}...[/bold cyan]\n")
+
+        try:
+            deploy_parallel_functions(
+                account=account,
+                user=user,
+                password=password,
+                warehouse=warehouse,
+                role=role,
+                parallel_api_key=api_key,
+                authenticator=authenticator if not password else None,
+                passcode=passcode,
+            )
+            console.print("\n[bold green]Deployment complete![/bold green]")
+            console.print("\n[cyan]Example query:[/cyan]")
+            console.print("""
+-- Basic usage (returns JSON with enriched fields and basis/citations)
+SELECT PARALLEL_INTEGRATION.ENRICHMENT.parallel_enrich(
+    OBJECT_CONSTRUCT('company_name', 'Google', 'website', 'google.com'),
+    ARRAY_CONSTRUCT('CEO name', 'Founding year', 'Brief description')
+) AS enriched_data;
+
+-- Parsing the JSON result into columns
+SELECT
+    data:ceo_name::STRING AS ceo_name,
+    data:founding_year::STRING AS founding_year,
+    data:brief_description::STRING AS description,
+    data:basis AS basis
+FROM (
+    SELECT PARALLEL_INTEGRATION.ENRICHMENT.parallel_enrich(
+        OBJECT_CONSTRUCT('company_name', 'Google', 'website', 'google.com'),
+        ARRAY_CONSTRUCT('CEO name', 'Founding year', 'Brief description')
+    ) AS data
+);
+""")
+        except Exception as e:
+            console.print(f"[bold red]Deployment failed: {e}[/bold red]")
+            raise click.Abort() from None
+
 
 # =============================================================================
 # Research Command Group
diff --git a/parallel_web_tools/core/batch.py b/parallel_web_tools/core/batch.py
index 0d75e3f..2da6b52 100644
--- a/parallel_web_tools/core/batch.py
+++ b/parallel_web_tools/core/batch.py
@@ -32,45 +32,52 @@ def build_output_schema(output_columns: list[str]) -> dict[str, Any]:
     }
 
 
+def _parse_content(content) -> dict[str, Any]:
+    """Parse API response content into a dictionary."""
+    if isinstance(content, dict):
+        return dict(content)
+    if isinstance(content, str):
+        try:
+            return json.loads(content)
+        except json.JSONDecodeError:
+            return {"result": content}
+    return {"result": str(content)}
+
+
 def extract_basis(output) -> list[dict[str, Any]]:
     """Extract basis/citations from a Parallel API output."""
+    if not getattr(output, "basis", None):
+        return []
+
     basis_list: list[dict[str, Any]] = []
+    for field_basis in output.basis:
+        entry: dict[str, Any] = {}
 
-    if not hasattr(output, "basis") or not output.basis:
-        return basis_list
+        if field := getattr(field_basis, "field", None):
+            entry["field"] = field
 
-    for field_basis in output.basis:
-        basis_entry: dict[str, Any] = {}
-
-        if hasattr(field_basis, "field") and field_basis.field:
-            basis_entry["field"] = field_basis.field
-
-        if hasattr(field_basis, "citations") and field_basis.citations:
-            basis_entry["citations"] = [
-                {
-                    "url": c.url if hasattr(c, "url") else None,
-                    "excerpts": c.excerpts if hasattr(c, "excerpts") else [],
-                }
-                for c in field_basis.citations
+        if citations := getattr(field_basis, "citations", None):
+            entry["citations"] = [
+                {"url": getattr(c, "url", None), "excerpts": getattr(c, "excerpts", [])} for c in citations
             ]
 
-        if hasattr(field_basis, "reasoning") and field_basis.reasoning:
-            basis_entry["reasoning"] = field_basis.reasoning
+        if reasoning := getattr(field_basis, "reasoning", None):
+            entry["reasoning"] = reasoning
 
-        if hasattr(field_basis, "confidence") and field_basis.confidence:
-            basis_entry["confidence"] = field_basis.confidence
+        if confidence := getattr(field_basis, "confidence", None):
+            entry["confidence"] = confidence
 
         # Fallback for simpler basis format
-        if not basis_entry:
-            if hasattr(field_basis, "url") and field_basis.url:
-                basis_entry["url"] = field_basis.url
-            if hasattr(field_basis, "title") and field_basis.title:
-                basis_entry["title"] = field_basis.title
-            if hasattr(field_basis, "excerpts") and field_basis.excerpts:
-                basis_entry["excerpts"] = field_basis.excerpts
+        if not entry:
+            if url := getattr(field_basis, "url", None):
+                entry["url"] = url
+            if title := getattr(field_basis, "title", None):
+                entry["title"] = title
+            if excerpts := getattr(field_basis, "excerpts", None):
+                entry["excerpts"] = excerpts
 
-        if basis_entry:
-            basis_list.append(basis_entry)
+        if entry:
+            basis_list.append(entry)
 
     return basis_list
 
@@ -148,22 +155,10 @@ def enrich_batch(
         for event in runs_stream:
             if event.type == "task_run.state":
                 run_id = event.run.run_id
-                if event.output and hasattr(event.output, "content"):
-                    content = event.output.content
-                    result: dict[str, Any]
-                    if isinstance(content, dict):
-                        result = dict(content)
-                    elif isinstance(content, str):
-                        try:
-                            result = json.loads(content)
-                        except json.JSONDecodeError:
-                            result = {"result": content}
-                    else:
-                        result = {"result": str(content)}
-
+                if content := getattr(event.output, "content", None):
+                    result = _parse_content(content)
                     if include_basis:
                         result["basis"] = extract_basis(event.output)
-
                     results_by_id[run_id] = result
                 elif event.run.error:
                     results_by_id[run_id] = {"error": str(event.run.error)}
@@ -202,111 +197,85 @@ def run_tasks(
 ) -> list[Any]:
     """Run batch tasks using Pydantic models for schema.
 
-    This is the async-based batch processing using task groups.
-    For simpler use cases, use enrich_batch() instead.
+    Uses the Parallel SDK's task group API with proper SSE handling.
     """
-    import asyncio
     import logging
     import uuid
     from datetime import UTC, datetime
 
-    from parallel.types import TaskSpecParam
+    from parallel import Parallel
+    from parallel.types import JsonSchemaParam, TaskSpecParam
+    from parallel.types.beta import BetaRunInputParam
+
+    from parallel_web_tools.core.auth import resolve_api_key
 
     logger = logging.getLogger(__name__)
 
-    def build_task_spec_param(input_schema, output_schema) -> TaskSpecParam:
-        return {
-            "input_schema": {"type": "json", "json_schema": input_schema.model_json_schema()},
-            "output_schema": {"type": "json", "json_schema": output_schema.model_json_schema()},
-        }
-
-    async def run_batch_task(
-        input_data: list[dict[str, Any]],
-        InputModel,
-        OutputModel,
-        processor: str,
-        batch_size: int = 100,
-    ):
-        import httpx
-
-        from parallel_web_tools.core.auth import get_api_key
-
-        batch_id = str(uuid.uuid4())
-        logger.info(f"Generated batch_id: {batch_id}")
-
-        api_key = get_api_key()
-        base_url = "https://api.parallel.ai"
-        headers = {
-            "Authorization": f"Bearer {api_key}",
-            "Content-Type": "application/json",
-            "Anthropic-Beta": "tasks-2025-01-15",
-        }
-
-        async with httpx.AsyncClient(base_url=base_url, headers=headers, timeout=120) as client:
-            # Create task group
-            response = await client.post("/v1beta/tasks/groups", json={})
-            response.raise_for_status()
-            group_response = response.json()
-            taskgroup_id = group_response["taskgroup_id"]
-            logger.info(f"Created taskgroup id {taskgroup_id}")
-
-            total_created = 0
-
-            for i in range(0, len(input_data), batch_size):
-                batch = input_data[i : i + batch_size]
-                run_inputs = [{"input": row, "processor": processor} for row in batch]
-                task_spec = build_task_spec_param(InputModel, OutputModel)
-
-                response = await client.post(
-                    f"/v1beta/tasks/groups/{taskgroup_id}/runs",
-                    json={"default_task_spec": task_spec, "inputs": run_inputs},
+    batch_id = str(uuid.uuid4())
+    logger.info(f"Generated batch_id: {batch_id}")
+
+    client = Parallel(api_key=resolve_api_key(None))
+
+    # Build task spec from Pydantic models
+    task_spec = TaskSpecParam(
+        input_schema=JsonSchemaParam(type="json", json_schema=InputModel.model_json_schema()),
+        output_schema=JsonSchemaParam(type="json", json_schema=OutputModel.model_json_schema()),
+    )
+
+    # Create task group
+    task_group = client.beta.task_group.create()
+    taskgroup_id = task_group.task_group_id
+    logger.info(f"Created taskgroup id {taskgroup_id}")
+
+    # Add runs in batches
+    batch_size = 100
+    total_created = 0
+    for i in range(0, len(input_data), batch_size):
+        batch = input_data[i : i + batch_size]
+        run_inputs: list[BetaRunInputParam] = [{"input": row, "processor": processor} for row in batch]
+        response = client.beta.task_group.add_runs(
+            taskgroup_id,
+            default_task_spec=task_spec,
+            inputs=run_inputs,
+        )
+        total_created += len(response.run_ids)
+        logger.info(f"Processing {i + len(batch)} entities. Created {total_created} Tasks.")
+
+    # Wait for completion
+    import time
+
+    time.sleep(3)  # Initial delay
+    while True:
+        status = client.beta.task_group.retrieve(taskgroup_id)
+        status_counts = status.status.task_run_status_counts or {}
+        logger.info(f"Status: {status_counts}")
+
+        if not status.status.is_active:
+            logger.info("All tasks completed!")
+            break
+
+        time.sleep(10)
+
+    # Get results using SDK's streaming (handles SSE properly)
+    results = []
+    runs_stream = client.beta.task_group.get_runs(taskgroup_id, include_input=True, include_output=True)
+
+    for event in runs_stream:
+        if event.type == "task_run.state" and event.output:
+            try:
+                input_val = InputModel.model_validate(event.input.input if event.input else {})
+                content = _parse_content(event.output.content)
+                output_val = OutputModel.model_validate(content)
+                results.append(
+                    {
+                        **input_val.model_dump(),
+                        **output_val.model_dump(),
+                        "batch_id": batch_id,
+                        "insertion_timestamp": datetime.now(UTC).isoformat(),
+                    }
                 )
-                response.raise_for_status()
-                resp_data = response.json()
-                total_created += len(resp_data.get("run_ids", []))
-                logger.info(f"Processing {i + len(batch)} entities. Created {total_created} Tasks.")
-
-            # Wait for completion
-            while True:
-                response = await client.get(f"/v1beta/tasks/groups/{taskgroup_id}")
-                response.raise_for_status()
-                resp_data = response.json()
-                status = resp_data.get("status", {})
-                logger.info(f"Status: {status.get('task_run_status_counts', {})}")
-
-                if not status.get("is_active", True):
-                    logger.info("All tasks completed!")
-                    break
-
-                await asyncio.sleep(10)
-
-            # Get results - use streaming endpoint
-            results = []
-            path = f"/v1beta/tasks/groups/{taskgroup_id}/runs?include_input=true&include_output=true"
-
-            async with client.stream("GET", path) as response:
-                response.raise_for_status()
-                async for line in response.aiter_lines():
-                    if not line.strip():
-                        continue
-                    # Parse SSE data lines
-                    if line.startswith("data: "):
-                        import json
-
-                        event = json.loads(line[6:])
-                        if event.get("type") == "task_run.state" and event.get("output"):
-                            input_val = InputModel.model_validate(event["input"]["input"])
-                            output_val = OutputModel.model_validate(event["output"]["content"])
-                            results.append(
-                                {
-                                    **input_val.model_dump(),
-                                    **output_val.model_dump(),
-                                    "batch_id": batch_id,
-                                    "insertion_timestamp": datetime.now(UTC).isoformat(),
-                                }
-                            )
-
-            logger.info(f"Successfully processed {len(results)} entities.")
-            return results
-
-    return asyncio.run(run_batch_task(input_data, InputModel, OutputModel, processor))
+            except Exception as e:
+                logger.warning(f"Failed to parse result: {e}")
+
+    logger.info(f"Successfully processed {len(results)} entities.")
+    return results
diff --git a/parallel_web_tools/integrations/snowflake/__init__.py b/parallel_web_tools/integrations/snowflake/__init__.py
index a1acb27..2cc8cef 100644
--- a/parallel_web_tools/integrations/snowflake/__init__.py
+++ b/parallel_web_tools/integrations/snowflake/__init__.py
@@ -17,7 +17,7 @@
     )
 
 SQL Usage:
-    SELECT parallel_enrich(
+    SELECT PARALLEL_INTEGRATION.ENRICHMENT.parallel_enrich(
         OBJECT_CONSTRUCT('company_name', 'Google'),
         ARRAY_CONSTRUCT('CEO name', 'Founding year')
     ) AS enriched_data;
diff --git a/parallel_web_tools/integrations/snowflake/deploy.py b/parallel_web_tools/integrations/snowflake/deploy.py
index 13fb51f..a392db4 100644
--- a/parallel_web_tools/integrations/snowflake/deploy.py
+++ b/parallel_web_tools/integrations/snowflake/deploy.py
@@ -87,50 +87,101 @@ def get_cleanup_sql() -> str:
     return get_sql_template("03_cleanup")
 
 
-def _check_existing_resources(cursor, database: str, schema: str) -> list[str]:
-    """Check which Snowflake resources already exist and would be overwritten."""
-    existing = []
-
-    # Check database
+def _check_resource_exists(cursor, query: str) -> bool:
+    """Execute a SHOW query and return True if any result exists."""
     try:
-        cursor.execute(f"SHOW DATABASES LIKE '{database}'")
-        if cursor.fetchone():
-            existing.append(f"Database: {database}")
+        cursor.execute(query)
+        return cursor.fetchone() is not None
     except Exception:
-        pass
+        return False
+
+
+def _strip_sql_comments(sql_text: str) -> str:
+    """Strip SQL comment lines from a statement."""
+    lines = [line for line in sql_text.split("\n") if not line.strip().startswith("--")]
+    return "\n".join(lines).strip()
+
 
-    # Check schema (only if database exists)
-    if existing:
+def _is_critical_statement(sql: str) -> bool:
+    """Check if a SQL statement is critical (should fail deployment if it errors)."""
+    return "CREATE" in sql.upper()
+
+
+def _execute_sql_statements(cursor, sql: str) -> list[str]:
+    """Execute semicolon-separated SQL statements, collecting critical errors.
+
+    Returns list of error messages for critical statements that failed.
+    """
+    errors = []
+    for statement in sql.split(";"):
+        clean = _strip_sql_comments(statement)
+        if not clean:
+            continue
         try:
-            cursor.execute(f"SHOW SCHEMAS LIKE '{schema}' IN DATABASE {database}")
-            if cursor.fetchone():
-                existing.append(f"Schema: {database}.{schema}")
-        except Exception:
-            pass
+            cursor.execute(clean)
+        except Exception as e:
+            clean_upper = clean.upper()
+            if _is_critical_statement(clean):
+                errors.append(str(e))
+                print(f"Error: {e}")
+            elif "SHOW" not in clean_upper and "SELECT" not in clean_upper:
+                print(f"Warning: {e}")
+    return errors
+
+
+def _build_connection_params(
+    account: str,
+    user: str,
+    warehouse: str,
+    role: str,
+    password: str | None = None,
+    authenticator: str | None = None,
+    passcode: str | None = None,
+    database: str | None = None,
+    schema: str | None = None,
+) -> dict:
+    """Build Snowflake connection parameters."""
+    params = {"account": account, "user": user, "warehouse": warehouse, "role": role}
+
+    if database:
+        params["database"] = database
+    if schema:
+        params["schema"] = schema
 
-    # Check external access integration
-    try:
-        cursor.execute("SHOW EXTERNAL ACCESS INTEGRATIONS LIKE 'parallel_api_access_integration'")
-        if cursor.fetchone():
-            existing.append("External Access Integration: parallel_api_access_integration")
-    except Exception:
-        pass
+    if password:
+        params["password"] = password
+        if authenticator:
+            params["authenticator"] = authenticator
+        if passcode:
+            params["passcode"] = passcode
+    elif authenticator:
+        params["authenticator"] = authenticator
+    else:
+        params["authenticator"] = "externalbrowser"
 
-    # Check secret
-    try:
-        cursor.execute(f"SHOW SECRETS LIKE 'parallel_api_key' IN DATABASE {database}")
-        if cursor.fetchone():
-            existing.append(f"Secret: {database}.{schema}.parallel_api_key")
-    except Exception:
-        pass
+    return params
 
-    # Check network rule
-    try:
-        cursor.execute(f"SHOW NETWORK RULES LIKE 'parallel_api_network_rule' IN DATABASE {database}")
-        if cursor.fetchone():
+
+def _check_existing_resources(cursor, database: str, schema: str) -> list[str]:
+    """Check which Snowflake resources already exist and would be overwritten."""
+    existing = []
+
+    db_exists = _check_resource_exists(cursor, f"SHOW DATABASES LIKE '{database}'")
+    if db_exists:
+        existing.append(f"Database: {database}")
+
+    if _check_resource_exists(cursor, "SHOW EXTERNAL ACCESS INTEGRATIONS LIKE 'parallel_api_access_integration'"):
+        existing.append("External Access Integration: parallel_api_access_integration")
+
+    if db_exists:
+        if _check_resource_exists(cursor, f"SHOW SCHEMAS LIKE '{schema}' IN DATABASE {database}"):
+            existing.append(f"Schema: {database}.{schema}")
+        if _check_resource_exists(cursor, f"SHOW SECRETS LIKE 'parallel_api_key' IN DATABASE {database}"):
+            existing.append(f"Secret: {database}.{schema}.parallel_api_key")
+        if _check_resource_exists(
+            cursor, f"SHOW NETWORK RULES LIKE 'parallel_api_network_rule' IN DATABASE {database}"
+        ):
             existing.append(f"Network Rule: {database}.{schema}.parallel_api_network_rule")
-    except Exception:
-        pass
 
     return existing
 
@@ -145,6 +196,7 @@ def deploy_parallel_functions(
     role: str = "ACCOUNTADMIN",
     parallel_api_key: str | None = None,
     authenticator: str | None = None,
+    passcode: str | None = None,
     force: bool = False,
 ) -> None:
     """
@@ -199,22 +251,15 @@ def deploy_parallel_functions(
             "or PARALLEL_API_KEY environment variable."
         )
 
-    # Build connection parameters
-    conn_params = {
-        "account": account,
-        "user": user,
-        "warehouse": warehouse,
-        "database": database,
-        "schema": schema,
-        "role": role,
-    }
-
-    if password:
-        conn_params["password"] = password
-    elif authenticator:
-        conn_params["authenticator"] = authenticator
-    else:
-        conn_params["authenticator"] = "externalbrowser"
+    conn_params = _build_connection_params(
+        account=account,
+        user=user,
+        warehouse=warehouse,
+        role=role,
+        password=password,
+        authenticator=authenticator,
+        passcode=passcode,
+    )
 
     # Connect to Snowflake
     print(f"Connecting to Snowflake account: {account}")
@@ -233,37 +278,15 @@ def deploy_parallel_functions(
 
         # Run setup SQL
         print("Running setup SQL (network rule, secret, integration)...")
-        setup_sql = get_setup_sql(api_key)
-        for statement in setup_sql.split(";"):
-            statement = statement.strip()
-            if statement and not statement.startswith("--"):
-                try:
-                    cursor.execute(statement)
-                except Exception as e:
-                    # Skip errors for verification queries
-                    if "SHOW" not in statement and "SELECT" not in statement:
-                        print(f"Warning: {e}")
+        setup_errors = _execute_sql_statements(cursor, get_setup_sql(api_key))
+        if setup_errors:
+            raise RuntimeError(f"Setup failed with {len(setup_errors)} error(s). See messages above.")
 
         # Run UDF creation SQL
         print("Creating parallel_enrich() UDF...")
-        udf_sql = get_udf_sql()
-        for statement in udf_sql.split(";"):
-            statement = statement.strip()
-            if statement and not statement.startswith("--"):
-                try:
-                    cursor.execute(statement)
-                except Exception as e:
-                    # Skip errors for verification queries
-                    if "SELECT" not in statement:
-                        print(f"Warning: {e}")
-
-        print("Deployment complete!")
-        print()
-        print("Test the integration with:")
-        print("  SELECT parallel_enrich(")
-        print("      OBJECT_CONSTRUCT('company_name', 'Google'),")
-        print("      ARRAY_CONSTRUCT('CEO name', 'Founding year')")
-        print("  ) AS enriched_data;")
+        udf_errors = _execute_sql_statements(cursor, get_udf_sql())
+        if udf_errors:
+            raise RuntimeError(f"UDF creation failed with {len(udf_errors)} error(s). See messages above.")
 
     finally:
         conn.close()
@@ -306,38 +329,29 @@ def cleanup_parallel_functions(
             "snowflake-connector-python is required. Install it with: pip install parallel-web-tools[snowflake]"
         ) from e
 
-    # Build connection parameters
-    conn_params = {
-        "account": account,
-        "user": user,
-        "warehouse": warehouse,
-        "database": "PARALLEL_INTEGRATION",
-        "schema": "ENRICHMENT",
-        "role": role,
-    }
-
-    if password:
-        conn_params["password"] = password
-    elif authenticator:
-        conn_params["authenticator"] = authenticator
-    else:
-        conn_params["authenticator"] = "externalbrowser"
+    conn_params = _build_connection_params(
+        account=account,
+        user=user,
+        warehouse=warehouse,
+        role=role,
+        password=password,
+        authenticator=authenticator,
+        database="PARALLEL_INTEGRATION",
+        schema="ENRICHMENT",
+    )
 
-    # Connect to Snowflake
     print(f"Connecting to Snowflake account: {account}")
     conn = snowflake.connector.connect(**conn_params)
 
     try:
         cursor = conn.cursor()
-
-        # Run cleanup SQL
         print("Running cleanup SQL...")
-        cleanup_sql = get_cleanup_sql()
-        for statement in cleanup_sql.split(";"):
-            statement = statement.strip()
-            if statement and not statement.startswith("--"):
+
+        for statement in get_cleanup_sql().split(";"):
+            clean = _strip_sql_comments(statement)
+            if clean:
                 try:
-                    cursor.execute(statement)
+                    cursor.execute(clean)
                 except Exception as e:
                     print(f"Warning: {e}")
 
diff --git a/parallel_web_tools/integrations/snowflake/sql/01_setup.sql b/parallel_web_tools/integrations/snowflake/sql/01_setup.sql
index 0b92d76..3785692 100644
--- a/parallel_web_tools/integrations/snowflake/sql/01_setup.sql
+++ b/parallel_web_tools/integrations/snowflake/sql/01_setup.sql
@@ -56,7 +56,14 @@ CREATE OR REPLACE EXTERNAL ACCESS INTEGRATION parallel_api_access_integration
     COMMENT = 'External access integration for Parallel API';
 
 -- =============================================================================
--- Step 4: Create Roles
+-- Step 4: Grant PyPI Repository Access
+-- =============================================================================
+-- Required for UDFs to use parallel-web-tools package from PyPI
+
+GRANT DATABASE ROLE SNOWFLAKE.PYPI_REPOSITORY_USER TO ROLE ACCOUNTADMIN;
+
+-- =============================================================================
+-- Step 5: Create Roles
 -- =============================================================================
 -- PARALLEL_DEVELOPER: Can create and modify UDFs
 -- PARALLEL_USER: Can execute UDFs
@@ -64,6 +71,9 @@ CREATE OR REPLACE EXTERNAL ACCESS INTEGRATION parallel_api_access_integration
 CREATE ROLE IF NOT EXISTS PARALLEL_DEVELOPER;
 CREATE ROLE IF NOT EXISTS PARALLEL_USER;
 
+-- Grant PyPI access to developer role
+GRANT DATABASE ROLE SNOWFLAKE.PYPI_REPOSITORY_USER TO ROLE PARALLEL_DEVELOPER;
+
 -- Grant permissions to PARALLEL_DEVELOPER
 GRANT USAGE ON DATABASE PARALLEL_INTEGRATION TO ROLE PARALLEL_DEVELOPER;
 GRANT USAGE ON SCHEMA PARALLEL_INTEGRATION.ENRICHMENT TO ROLE PARALLEL_DEVELOPER;
diff --git a/parallel_web_tools/integrations/snowflake/sql/02_create_udf.sql b/parallel_web_tools/integrations/snowflake/sql/02_create_udf.sql
index 368c00d..a02f030 100644
--- a/parallel_web_tools/integrations/snowflake/sql/02_create_udf.sql
+++ b/parallel_web_tools/integrations/snowflake/sql/02_create_udf.sql
@@ -6,11 +6,12 @@
 -- Prerequisites:
 -- - Run 01_setup.sql first to create network rule, secret, and integration
 -- - PARALLEL_DEVELOPER role or ACCOUNTADMIN
+-- - SNOWFLAKE.PYPI_REPOSITORY_USER role granted (for PyPI package access)
 --
 -- Usage:
 -- After running this script, you can use parallel_enrich() in SQL queries:
 --
---   SELECT parallel_enrich(
+--   SELECT PARALLEL_INTEGRATION.ENRICHMENT.parallel_enrich(
 --       OBJECT_CONSTRUCT('company_name', 'Google'),
 --       ARRAY_CONSTRUCT('CEO name', 'Founding year')
 --   ) AS enriched_data;
@@ -22,7 +23,8 @@ USE SCHEMA ENRICHMENT;
 -- =============================================================================
 -- Internal UDF (with API key parameter)
 -- =============================================================================
--- This is the internal implementation. Users should call the public wrapper.
+-- This is the internal implementation using parallel-web-tools from PyPI.
+-- It shares the same core enrichment logic as BigQuery and Spark integrations.
 
 CREATE OR REPLACE FUNCTION parallel_enrich_internal(
     input_data OBJECT,
@@ -32,184 +34,44 @@ CREATE OR REPLACE FUNCTION parallel_enrich_internal(
 )
 RETURNS VARIANT
 LANGUAGE PYTHON
-RUNTIME_VERSION = '3.11'
-PACKAGES = ('requests')
+RUNTIME_VERSION = '3.12'
+ARTIFACT_REPOSITORY = snowflake.snowpark.pypi_shared_repository
+PACKAGES = ('parallel-web-tools')
 HANDLER = 'enrich'
 EXTERNAL_ACCESS_INTEGRATIONS = (parallel_api_access_integration)
 SECRETS = ('api_key' = parallel_api_key)
 AS $$
 import _snowflake
-import json
-import time
-import requests
+from parallel_web_tools.core import enrich_batch
 
-def enrich(input_data: dict, output_columns: list, processor: str, api_key_override: str) -> dict:
-    """
-    Enrich data using the Parallel API.
-
-    Args:
-        input_data: Dictionary of input data (e.g., {"company_name": "Google"})
-        output_columns: List of output column descriptions (e.g., ["CEO name", "Founding year"])
-        processor: Parallel processor to use (e.g., "lite-fast", "base-fast")
-        api_key_override: Optional API key override (empty string to use secret)
 
-    Returns:
-        Dictionary with enriched data or error
-    """
+def enrich(input_data: dict, output_columns: list, processor: str, api_key_override: str) -> dict:
     # Get API key from secret or override
     if api_key_override:
         api_key = api_key_override
     else:
-        api_key = _snowflake.get_generic_secret_string('api_key')
+        api_key = _snowflake.get_generic_secret_string("api_key")
 
     if not api_key:
         return {"error": "No API key provided"}
 
-    # Build output schema from column descriptions
-    output_properties = {}
-    for col in output_columns:
-        # Extract base name (before parentheses, brackets, etc.)
-        base_name = col.split("(")[0].split("[")[0].strip()
-
-        # Convert to valid property name
-        prop_name = base_name.lower().replace(" ", "_").replace("-", "_")
-        prop_name = "".join(c for c in prop_name if c.isalnum() or c == "_")
-
-        # Add prefix if starts with number
-        if prop_name and prop_name[0].isdigit():
-            prop_name = "col_" + prop_name
-
-        if not prop_name:
-            prop_name = "column"
-
-        output_properties[prop_name] = {
-            "type": "string",
-            "description": col
-        }
-
-    output_schema = {
-        "type": "json",
-        "json_schema": {
-            "type": "object",
-            "properties": output_properties,
-            "required": list(output_properties.keys())
-        }
-    }
-
-    # API configuration
-    base_url = "https://api.parallel.ai"
-    headers = {
-        "x-api-key": api_key,
-        "Content-Type": "application/json",
-        "User-Agent": "Parallel-Snowflake-Integration/1.0"
-    }
-
     try:
-        # Step 1: Create task group
-        create_response = requests.post(
-            f"{base_url}/v1beta/tasks/groups",
-            headers=headers,
-            json={},
-            timeout=30
+        # Use shared core enrichment logic (same as BigQuery/Spark)
+        results = enrich_batch(
+            inputs=[input_data],
+            output_columns=list(output_columns),
+            api_key=api_key,
+            processor=processor,
+            timeout=300,
+            include_basis=True,
         )
-        create_response.raise_for_status()
-
-        task_group = create_response.json()
-        taskgroup_id = task_group.get("task_group_id")
-
-        if not taskgroup_id:
-            return {"error": "Failed to create task group", "response": str(task_group)}
-
-        # Step 2: Add run with input data
-        run_input = {
-            "default_task_spec": {
-                "output_schema": output_schema
-            },
-            "inputs": [
-                {"input": input_data, "processor": processor}
-            ]
-        }
-
-        add_response = requests.post(
-            f"{base_url}/v1beta/tasks/groups/{taskgroup_id}/runs",
-            headers=headers,
-            json=run_input,
-            timeout=30
-        )
-        add_response.raise_for_status()
-
-        # Step 3: Poll for completion (5 minute timeout)
-        max_wait = 300
-        start_time = time.time()
-
-        while time.time() - start_time < max_wait:
-            status_response = requests.get(
-                f"{base_url}/v1beta/tasks/groups/{taskgroup_id}",
-                headers=headers,
-                timeout=30
-            )
-            status_response.raise_for_status()
-
-            status_data = status_response.json()
-            counts = status_data.get("status", {}).get("task_run_status_counts", {})
-            completed = counts.get("completed", 0)
-            failed = counts.get("failed", 0)
-            total = status_data.get("status", {}).get("num_task_runs", 1)
-
-            if completed + failed >= total:
-                break
-
-            time.sleep(2)
-        else:
-            return {"error": "Timeout waiting for enrichment to complete"}
-
-        # Step 4: Get results via streaming endpoint
-        results_response = requests.get(
-            f"{base_url}/v1beta/tasks/groups/{taskgroup_id}/runs",
-            headers=headers,
-            params={"include_input": "true", "include_output": "true"},
-            timeout=120,
-            stream=True
-        )
-        results_response.raise_for_status()
-
-        # Parse Server-Sent Events
-        for line in results_response.iter_lines():
-            if line and line.startswith(b'data: '):
-                try:
-                    event_data = json.loads(line.decode('utf-8')[6:])
-                    if event_data.get("type") == "task_run.state":
-                        output = event_data.get("output", {})
-                        content = output.get("content")
-
-                        # Parse content (may be string or dict)
-                        if isinstance(content, str):
-                            try:
-                                result = json.loads(content)
-                            except json.JSONDecodeError:
-                                result = {"result": content}
-                        elif isinstance(content, dict):
-                            result = content
-                        else:
-                            result = {"result": str(content)}
-
-                        # Add basis (citations) if available
-                        basis = output.get("basis", [])
-                        if basis:
-                            result["basis"] = basis
-
-                        return result
-                except json.JSONDecodeError:
-                    continue
 
+        if results and len(results) > 0:
+            return results[0]
         return {"error": "No results received"}
 
-    except requests.exceptions.Timeout:
-        return {"error": "Request timeout"}
-    except requests.exceptions.RequestException as e:
-        return {"error": f"API request failed: {str(e)}"}
     except Exception as e:
-        return {"error": f"Unexpected error: {str(e)}"}
+        return {"error": f"Enrichment failed: {str(e)}"}
 $$;
 
 -- =============================================================================
@@ -251,10 +113,4 @@ GRANT USAGE ON FUNCTION parallel_enrich(OBJECT, ARRAY, VARCHAR) TO ROLE PARALLEL
 -- Verification
 -- =============================================================================
 
--- Test the function (requires valid API key in secret)
--- SELECT parallel_enrich(
---     OBJECT_CONSTRUCT('company_name', 'Google'),
---     ARRAY_CONSTRUCT('CEO name', 'Founding year')
--- ) AS enriched_data;
-
-SELECT 'UDF created successfully! Test with: SELECT parallel_enrich(OBJECT_CONSTRUCT(''company_name'', ''Google''), ARRAY_CONSTRUCT(''CEO name''))' AS status;
+SELECT 'parallel_enrich() UDF created successfully' AS status;

From fc65ad93660e7ada97a705c887c7dd3725b9dceb Mon Sep 17 00:00:00 2001
From: Matt Harris <mharris@parallel.ai>
Date: Sat, 24 Jan 2026 02:04:01 -0500
Subject: [PATCH 02/13] fix: use pyrefly config excludes in CI

---
 .github/workflows/ci.yml | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index cf179da..883c558 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -64,8 +64,7 @@ jobs:
         run: uv run ruff format --check parallel_web_tools/ tests/
 
       - name: Type check with pyrefly
-        run: uv run pyrefly check parallel_web_tools/
-        continue-on-error: true
+        run: uv run pyrefly check
 
   cli-test:
     runs-on: ubuntu-latest

From fda5f4bb7327d6606872dd4fb0f1d7a7a3b9aff7 Mon Sep 17 00:00:00 2001
From: Matt Harris <mharris@parallel.ai>
Date: Sat, 24 Jan 2026 02:09:51 -0500
Subject: [PATCH 03/13] fix: update run_tasks test to mock Parallel SDK instead
 of httpx

---
 parallel_web_tools/core/batch.py |   3 +-
 tests/test_enrichment.py         | 118 ++++++++++---------------------
 2 files changed, 40 insertions(+), 81 deletions(-)

diff --git a/parallel_web_tools/core/batch.py b/parallel_web_tools/core/batch.py
index 2da6b52..c3a1afc 100644
--- a/parallel_web_tools/core/batch.py
+++ b/parallel_web_tools/core/batch.py
@@ -244,7 +244,6 @@ def run_tasks(
     # Wait for completion
     import time
 
-    time.sleep(3)  # Initial delay
     while True:
         status = client.beta.task_group.retrieve(taskgroup_id)
         status_counts = status.status.task_run_status_counts or {}
@@ -254,7 +253,7 @@ def run_tasks(
             logger.info("All tasks completed!")
             break
 
-        time.sleep(10)
+        time.sleep(2)
 
     # Get results using SDK's streaming (handles SSE properly)
     results = []
diff --git a/tests/test_enrichment.py b/tests/test_enrichment.py
index 8087286..9a9a3f5 100644
--- a/tests/test_enrichment.py
+++ b/tests/test_enrichment.py
@@ -792,10 +792,10 @@ def test_default_parameters(self):
 
 
 class TestRunTasks:
-    """Tests for run_tasks function with httpx-based API calls."""
+    """Tests for run_tasks function with Parallel SDK."""
 
     def test_run_tasks_basic(self):
-        """Should process batch tasks using httpx and return results."""
+        """Should process batch tasks using the Parallel SDK and return results."""
         from pydantic import BaseModel
 
         class InputModel(BaseModel):
@@ -804,88 +804,48 @@ class InputModel(BaseModel):
         class OutputModel(BaseModel):
             ceo: str
 
-        # Mock responses for the API flow
-        mock_responses = [
-            # Create task group
-            mock.MagicMock(status_code=200, json=mock.MagicMock(return_value={"taskgroup_id": "tgrp_123"})),
-            # Add runs
-            mock.MagicMock(status_code=200, json=mock.MagicMock(return_value={"run_ids": ["run_1", "run_2"]})),
-            # Check status - still running
-            mock.MagicMock(
-                status_code=200,
-                json=mock.MagicMock(
-                    return_value={"status": {"is_active": True, "task_run_status_counts": {"running": 2}}}
-                ),
-            ),
-            # Check status - completed
-            mock.MagicMock(
-                status_code=200,
-                json=mock.MagicMock(
-                    return_value={"status": {"is_active": False, "task_run_status_counts": {"completed": 2}}}
-                ),
-            ),
-        ]
-
-        # Mock the streaming response
-        class MockStreamResponse:
-            def __init__(self):
-                self.status_code = 200
-
-            async def __aenter__(self):
-                return self
-
-            async def __aexit__(self, *args):
-                pass
-
-            def raise_for_status(self):
-                pass
-
-            async def aiter_lines(self):
-                yield 'data: {"type": "task_run.state", "input": {"input": {"company": "Anthropic"}}, "output": {"content": {"ceo": "Dario Amodei"}}}'
-                yield 'data: {"type": "task_run.state", "input": {"input": {"company": "OpenAI"}}, "output": {"content": {"ceo": "Sam Altman"}}}'
-
-        with mock.patch("parallel_web_tools.core.auth.get_api_key", return_value="test-key"):
-            with mock.patch("httpx.AsyncClient") as mock_client_class:
-                mock_client = mock.MagicMock()
-
-                # Set up response iterator
-                response_iter = iter(mock_responses)
-
-                async def mock_post(*args, **kwargs):
-                    resp = next(response_iter)
-                    resp.raise_for_status = mock.MagicMock()
-                    return resp
+        # Mock the Parallel SDK client
+        mock_client = mock.MagicMock()
 
-                async def mock_get(*args, **kwargs):
-                    resp = next(response_iter)
-                    resp.raise_for_status = mock.MagicMock()
-                    return resp
+        # Mock task group create
+        mock_client.beta.task_group.create.return_value = mock.MagicMock(task_group_id="tgrp_123")
 
-                mock_client.post = mock_post
-                mock_client.get = mock_get
-                mock_client.stream = mock.MagicMock(return_value=MockStreamResponse())
+        # Mock add_runs
+        mock_client.beta.task_group.add_runs.return_value = mock.MagicMock(run_ids=["run_1", "run_2"])
 
-                async def async_context_manager(*args, **kwargs):
-                    return mock_client
+        # Mock retrieve (status check) - return completed immediately
+        mock_client.beta.task_group.retrieve.return_value = mock.MagicMock(
+            status=mock.MagicMock(is_active=False, task_run_status_counts={"completed": 2})
+        )
 
-                mock_client.__aenter__ = async_context_manager
-                mock_client.__aexit__ = mock.AsyncMock(return_value=None)
-                mock_client_class.return_value = mock_client
+        # Mock get_runs (streaming results)
+        mock_event1 = mock.MagicMock(
+            type="task_run.state",
+            input=mock.MagicMock(input={"company": "Anthropic"}),
+            output=mock.MagicMock(content={"ceo": "Dario Amodei"}),
+        )
+        mock_event2 = mock.MagicMock(
+            type="task_run.state",
+            input=mock.MagicMock(input={"company": "OpenAI"}),
+            output=mock.MagicMock(content={"ceo": "Sam Altman"}),
+        )
+        mock_client.beta.task_group.get_runs.return_value = [mock_event1, mock_event2]
 
-                input_data = [
-                    {"company": "Anthropic"},
-                    {"company": "OpenAI"},
-                ]
+        with mock.patch("parallel_web_tools.core.auth.resolve_api_key", return_value="test-key"):
+            with mock.patch("parallel.Parallel", return_value=mock_client):
+                with mock.patch("time.sleep"):  # Speed up test
+                    input_data = [
+                        {"company": "Anthropic"},
+                        {"company": "OpenAI"},
+                    ]
 
-                # Patch asyncio.sleep to speed up test
-                with mock.patch("asyncio.sleep", return_value=None):
                     results = run_tasks(input_data, InputModel, OutputModel, "lite-fast")
 
-                assert len(results) == 2
-                assert results[0]["company"] == "Anthropic"
-                assert results[0]["ceo"] == "Dario Amodei"
-                assert results[1]["company"] == "OpenAI"
-                assert results[1]["ceo"] == "Sam Altman"
-                # Check batch_id and timestamp are added
-                assert "batch_id" in results[0]
-                assert "insertion_timestamp" in results[0]
+        assert len(results) == 2
+        assert results[0]["company"] == "Anthropic"
+        assert results[0]["ceo"] == "Dario Amodei"
+        assert results[1]["company"] == "OpenAI"
+        assert results[1]["ceo"] == "Sam Altman"
+        # Check batch_id and timestamp are added
+        assert "batch_id" in results[0]
+        assert "insertion_timestamp" in results[0]

From 50d58e2745b0d3048e222acd0a2eeec06247a124 Mon Sep 17 00:00:00 2001
From: Matt Harris <mharris@parallel.ai>
Date: Sat, 24 Jan 2026 02:10:43 -0500
Subject: [PATCH 04/13] bump to 0.0.5

---
 parallel_web_tools/__init__.py                                  | 2 +-
 .../integrations/bigquery/cloud_function/requirements.txt       | 2 +-
 pyproject.toml                                                  | 2 +-
 tests/test_cli.py                                               | 2 +-
 uv.lock                                                         | 2 +-
 5 files changed, 5 insertions(+), 5 deletions(-)

diff --git a/parallel_web_tools/__init__.py b/parallel_web_tools/__init__.py
index ea07545..34c2aee 100644
--- a/parallel_web_tools/__init__.py
+++ b/parallel_web_tools/__init__.py
@@ -27,7 +27,7 @@
     run_tasks,
 )
 
-__version__ = "0.0.4"
+__version__ = "0.0.5"
 
 __all__ = [
     # Auth
diff --git a/parallel_web_tools/integrations/bigquery/cloud_function/requirements.txt b/parallel_web_tools/integrations/bigquery/cloud_function/requirements.txt
index fe3b239..e9ae780 100644
--- a/parallel_web_tools/integrations/bigquery/cloud_function/requirements.txt
+++ b/parallel_web_tools/integrations/bigquery/cloud_function/requirements.txt
@@ -1,5 +1,5 @@
 # Cloud Function dependencies for BigQuery Remote Function
 functions-framework>=3.0.0
 flask>=3.0.0
-parallel-web-tools>=0.0.4
+parallel-web-tools>=0.0.5
 google-cloud-secret-manager>=2.20.0
diff --git a/pyproject.toml b/pyproject.toml
index 4441f8b..f6bbecf 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "parallel-web-tools"
-version = "0.0.4"
+version = "0.0.5"
 description = "Parallel Tools: CLI and data enrichment utilities for the Parallel API"
 readme = "README.md"
 requires-python = ">=3.12"
diff --git a/tests/test_cli.py b/tests/test_cli.py
index ceb00fb..95aab67 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -225,7 +225,7 @@ def test_version(self, runner):
         """Should show version."""
         result = runner.invoke(main, ["--version"])
         assert result.exit_code == 0
-        assert "0.0.4" in result.output
+        assert "0.0.5" in result.output
 
 
 class TestAuthCommand:
diff --git a/uv.lock b/uv.lock
index 29ab4ef..903b1f9 100644
--- a/uv.lock
+++ b/uv.lock
@@ -1057,7 +1057,7 @@ wheels = [
 
 [[package]]
 name = "parallel-web-tools"
-version = "0.0.4"
+version = "0.0.5"
 source = { editable = "." }
 dependencies = [
     { name = "pandas" },

From 5b2daa8ba5c44b8c52029329ff45021528f1fc3d Mon Sep 17 00:00:00 2001
From: Matt Harris <mharris@parallel.ai>
Date: Sat, 24 Jan 2026 02:23:33 -0500
Subject: [PATCH 05/13] ci: retrigger tests


From 2d4f225f4ac82fc94bced67402c43bc31044fe2d Mon Sep 17 00:00:00 2001
From: Matt Harris <mharris@parallel.ai>
Date: Sat, 24 Jan 2026 02:30:08 -0500
Subject: [PATCH 06/13] fix: validate deploy params before triggering OAuth

Move parameter validation (--project for BigQuery, --account/--user for
Snowflake) before API key resolution. This prevents triggering the OAuth
flow when required parameters are missing, improving UX and fixing CI
tests that were hanging when invoking deploy without required params.
---
 parallel_web_tools/cli/commands.py | 40 ++++++++++++++----------------
 1 file changed, 18 insertions(+), 22 deletions(-)

diff --git a/parallel_web_tools/cli/commands.py b/parallel_web_tools/cli/commands.py
index 968c067..e6656f6 100644
--- a/parallel_web_tools/cli/commands.py
+++ b/parallel_web_tools/cli/commands.py
@@ -758,24 +758,26 @@ def enrich_deploy(
     role: str,
 ):
     """Deploy Parallel enrichment to a cloud system."""
-    # Resolve API key for all systems
-    if not api_key:
-        api_key = os.environ.get("PARALLEL_API_KEY")
-    if not api_key:
-        try:
-            api_key = get_api_key()
-        except Exception:
-            pass
-    if not api_key:
-        console.print("[bold red]Error: Parallel API key required[/bold red]")
-        console.print("  Use --api-key, PARALLEL_API_KEY env var, or run 'parallel-cli login'")
-        raise click.Abort()
+    from parallel_web_tools.core.auth import get_api_key
 
-    if system == "bigquery":
-        if not project:
-            console.print("[bold red]Error: --project is required for BigQuery deployment.[/bold red]")
+    # Validate required parameters FIRST (before triggering OAuth)
+    if system == "bigquery" and not project:
+        console.print("[bold red]Error: --project is required for BigQuery deployment.[/bold red]")
+        raise click.Abort()
+    if system == "snowflake":
+        if not account:
+            console.print("[bold red]Error: --account is required for Snowflake deployment.[/bold red]")
+            raise click.Abort()
+        if not user:
+            console.print("[bold red]Error: --user is required for Snowflake deployment.[/bold red]")
             raise click.Abort()
 
+    # Now resolve API key (may trigger OAuth flow if needed)
+    if not api_key:
+        api_key = get_api_key()
+
+    if system == "bigquery":
+        assert project is not None  # Validated above
         from parallel_web_tools.integrations.bigquery import deploy_bigquery_integration
 
         console.print(f"[bold cyan]Deploying to BigQuery in {project}...[/bold cyan]\n")
@@ -796,13 +798,7 @@ def enrich_deploy(
             raise click.Abort() from None
 
     elif system == "snowflake":
-        if not account:
-            console.print("[bold red]Error: --account is required for Snowflake deployment.[/bold red]")
-            raise click.Abort()
-        if not user:
-            console.print("[bold red]Error: --user is required for Snowflake deployment.[/bold red]")
-            raise click.Abort()
-
+        assert account is not None and user is not None  # Validated above
         from parallel_web_tools.integrations.snowflake import deploy_parallel_functions
 
         console.print(f"[bold cyan]Deploying to Snowflake account {account}...[/bold cyan]\n")

From ec98fe81cd49489c4850d06faeedac4cdf450fbe Mon Sep 17 00:00:00 2001
From: Matt Harris <mharris@parallel.ai>
Date: Sat, 24 Jan 2026 20:30:37 -0500
Subject: [PATCH 07/13] refactor: remove deploy from standalone CLI, require
 pip install

- Remove bigquery/snowflake integrations from PyInstaller spec
- Add graceful error handling when deploy modules are not available
- Update documentation to clarify deploy requires pip install with extras
- Standalone CLI now focuses on core features (search, extract, enrich)
- Deploy commands require: pip install parallel-web-tools[snowflake|bigquery]
---
 README.md                          |  6 ++++--
 docs/bigquery-setup.md             |  2 ++
 docs/snowflake-setup.md            |  2 ++
 parallel-web-tools.spec            |  9 ++-------
 parallel_web_tools/cli/commands.py | 14 ++++++++++++--
 5 files changed, 22 insertions(+), 11 deletions(-)

diff --git a/README.md b/README.md
index 5af68da..30e07c2 100644
--- a/README.md
+++ b/README.md
@@ -20,7 +20,7 @@ CLI and data enrichment utilities for the [Parallel API](https://docs.parallel.a
 
 ### Standalone CLI (Recommended)
 
-Install the standalone `parallel-cli` binary with everything bundled (no Python required):
+Install the standalone `parallel-cli` binary for search, extract, and enrichment (no Python required):
 
 ```bash
 curl -fsSL https://raw.githubusercontent.com/parallel-web/parallel-web-tools/main/install-cli.sh | bash
@@ -28,6 +28,8 @@ curl -fsSL https://raw.githubusercontent.com/parallel-web/parallel-web-tools/mai
 
 This automatically detects your platform (macOS/Linux, x64/arm64) and installs to `~/.local/bin`.
 
+> **Note:** The standalone binary includes core CLI features. For deployment commands (`enrich deploy`), use pip: `pip install parallel-web-tools[snowflake]` or `[bigquery]`.
+
 ### Python Package
 
 For programmatic usage or data enrichment integrations:
@@ -60,7 +62,7 @@ parallel-cli
     ├── run                 # Run enrichment
     ├── plan                # Create YAML config
     ├── suggest             # AI suggests output columns
-    └── deploy              # Deploy to cloud systems (BigQuery, etc.)
+    └── deploy              # Deploy to cloud systems (requires pip install)
 ```
 
 ## Quick Start
diff --git a/docs/bigquery-setup.md b/docs/bigquery-setup.md
index 307aa60..36437f6 100644
--- a/docs/bigquery-setup.md
+++ b/docs/bigquery-setup.md
@@ -36,6 +36,8 @@ Parallel Task API
 pip install parallel-web-tools[bigquery]
 ```
 
+> **Note:** The standalone `parallel-cli` binary does not include deployment commands. You must install via pip with the `[bigquery]` extra to use `parallel-cli enrich deploy --system bigquery`.
+
 ## Quick Start Deployment
 
 ### Option 1: CLI (Recommended)
diff --git a/docs/snowflake-setup.md b/docs/snowflake-setup.md
index 8952df3..68d613d 100644
--- a/docs/snowflake-setup.md
+++ b/docs/snowflake-setup.md
@@ -78,6 +78,8 @@ Or with uv:
 uv add "parallel-web-tools[snowflake]"
 ```
 
+> **Note:** The standalone `parallel-cli` binary does not include deployment commands. You must install via pip/uv with the `[snowflake]` extra to use `parallel-cli enrich deploy --system snowflake`.
+
 ## Quick Start - CLI Deployment
 
 The easiest way to deploy is using the CLI:
diff --git a/parallel-web-tools.spec b/parallel-web-tools.spec
index 55950c3..756361c 100644
--- a/parallel-web-tools.spec
+++ b/parallel-web-tools.spec
@@ -41,13 +41,8 @@ a = Analysis(
         'parallel_web_tools.processors.csv',
         'parallel_web_tools.processors.duckdb',
         'parallel_web_tools.processors.bigquery',
-        # Integrations with deploy capability (bundled in CLI)
-        'parallel_web_tools.integrations',
-        'parallel_web_tools.integrations.bigquery',
-        'parallel_web_tools.integrations.bigquery.deploy',
-        'parallel_web_tools.integrations.snowflake',
-        'parallel_web_tools.integrations.snowflake.deploy',
-        # Note: polars, duckdb, spark integrations are library-only (no deploy step)
+        # Note: Deploy commands (bigquery, snowflake) are NOT included in standalone CLI
+        # They require: pip install parallel-web-tools[snowflake] or [bigquery]
         # Dependencies that might not be auto-detected
         'click',
         'questionary',
diff --git a/parallel_web_tools/cli/commands.py b/parallel_web_tools/cli/commands.py
index e6656f6..11804fe 100644
--- a/parallel_web_tools/cli/commands.py
+++ b/parallel_web_tools/cli/commands.py
@@ -778,7 +778,12 @@ def enrich_deploy(
 
     if system == "bigquery":
         assert project is not None  # Validated above
-        from parallel_web_tools.integrations.bigquery import deploy_bigquery_integration
+        try:
+            from parallel_web_tools.integrations.bigquery import deploy_bigquery_integration
+        except ImportError:
+            console.print("[bold red]Error: BigQuery deployment is not available in the standalone CLI.[/bold red]")
+            console.print("\nInstall via pip: [cyan]pip install parallel-web-tools[bigquery][/cyan]")
+            raise click.Abort() from None
 
         console.print(f"[bold cyan]Deploying to BigQuery in {project}...[/bold cyan]\n")
 
@@ -799,7 +804,12 @@ def enrich_deploy(
 
     elif system == "snowflake":
         assert account is not None and user is not None  # Validated above
-        from parallel_web_tools.integrations.snowflake import deploy_parallel_functions
+        try:
+            from parallel_web_tools.integrations.snowflake import deploy_parallel_functions
+        except ImportError:
+            console.print("[bold red]Error: Snowflake deployment is not available in the standalone CLI.[/bold red]")
+            console.print("\nInstall via pip: [cyan]pip install parallel-web-tools[snowflake][/cyan]")
+            raise click.Abort() from None
 
         console.print(f"[bold cyan]Deploying to Snowflake account {account}...[/bold cyan]\n")
 

From 496b92905a59142b9e2d79faeac64fa8c1ac624a Mon Sep 17 00:00:00 2001
From: Matt Harris <mharris@parallel.ai>
Date: Sat, 24 Jan 2026 20:51:59 -0500
Subject: [PATCH 08/13] docs: add deep research to standalone CLI description

---
 README.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/README.md b/README.md
index 30e07c2..4d99de1 100644
--- a/README.md
+++ b/README.md
@@ -20,7 +20,7 @@ CLI and data enrichment utilities for the [Parallel API](https://docs.parallel.a
 
 ### Standalone CLI (Recommended)
 
-Install the standalone `parallel-cli` binary for search, extract, and enrichment (no Python required):
+Install the standalone `parallel-cli` binary for search, extract, enrichment, and deep research (no Python required):
 
 ```bash
 curl -fsSL https://raw.githubusercontent.com/parallel-web/parallel-web-tools/main/install-cli.sh | bash

From c5ed89d6cffb51e0ed40fa6e85afb465e22c0aa4 Mon Sep 17 00:00:00 2001
From: Matt Harris <mharris@parallel.ai>
Date: Sat, 24 Jan 2026 20:55:46 -0500
Subject: [PATCH 09/13] fix: hide deploy command in standalone CLI

Conditionally register deploy command only when not running as frozen
executable (PyInstaller). This prevents showing a command in --help
that won't work in the standalone CLI.
---
 parallel_web_tools/cli/commands.py | 11 ++++++++++-
 1 file changed, 10 insertions(+), 1 deletion(-)

diff --git a/parallel_web_tools/cli/commands.py b/parallel_web_tools/cli/commands.py
index 11804fe..966da76 100644
--- a/parallel_web_tools/cli/commands.py
+++ b/parallel_web_tools/cli/commands.py
@@ -4,6 +4,7 @@
 import json
 import logging
 import os
+import sys
 import tempfile
 from typing import Any
 
@@ -728,7 +729,9 @@ def enrich_suggest(intent: str, source_columns: str | None, output_json: bool):
         raise click.Abort() from None
 
 
-@enrich.command(name="deploy")
+# Deploy command - only registered when not running as frozen executable (standalone CLI)
+# Standalone CLI users should use: pip install parallel-web-tools[snowflake|bigquery]
+@click.command(name="deploy")
 @click.option(
     "--system", type=click.Choice(["bigquery", "snowflake"]), required=True, help="Target system to deploy to"
 )
@@ -851,6 +854,12 @@ def enrich_deploy(
             raise click.Abort() from None
 
 
+# Only register deploy command when not running as frozen executable (PyInstaller)
+# Standalone CLI doesn't bundle deploy dependencies - use pip install instead
+if not getattr(sys, "frozen", False):
+    enrich.add_command(enrich_deploy)
+
+
 # =============================================================================
 # Research Command Group
 # =============================================================================

From d6dca05082b13820434b2a8077ca0678cf3f20e3 Mon Sep 17 00:00:00 2001
From: Matt Harris <mharris@parallel.ai>
Date: Sat, 24 Jan 2026 21:01:16 -0500
Subject: [PATCH 10/13] fix: correct BigQuery deploy install instructions

BigQuery deploy uses gcloud/bq CLI, not SQLAlchemy. The [bigquery] extra
is for SQLAlchemy-based access, not needed for deploy. Base package is
sufficient: pip install parallel-web-tools
---
 docs/bigquery-setup.md             | 4 ++--
 parallel_web_tools/cli/commands.py | 3 ++-
 2 files changed, 4 insertions(+), 3 deletions(-)

diff --git a/docs/bigquery-setup.md b/docs/bigquery-setup.md
index 36437f6..2594c81 100644
--- a/docs/bigquery-setup.md
+++ b/docs/bigquery-setup.md
@@ -33,10 +33,10 @@ Parallel Task API
 ## Installation
 
 ```bash
-pip install parallel-web-tools[bigquery]
+pip install parallel-web-tools
 ```
 
-> **Note:** The standalone `parallel-cli` binary does not include deployment commands. You must install via pip with the `[bigquery]` extra to use `parallel-cli enrich deploy --system bigquery`.
+> **Note:** The standalone `parallel-cli` binary does not include deployment commands. You must install via pip to use `parallel-cli enrich deploy --system bigquery`.
 
 ## Quick Start Deployment
 
diff --git a/parallel_web_tools/cli/commands.py b/parallel_web_tools/cli/commands.py
index 966da76..6ba67af 100644
--- a/parallel_web_tools/cli/commands.py
+++ b/parallel_web_tools/cli/commands.py
@@ -785,7 +785,8 @@ def enrich_deploy(
             from parallel_web_tools.integrations.bigquery import deploy_bigquery_integration
         except ImportError:
             console.print("[bold red]Error: BigQuery deployment is not available in the standalone CLI.[/bold red]")
-            console.print("\nInstall via pip: [cyan]pip install parallel-web-tools[bigquery][/cyan]")
+            console.print("\nInstall via pip: [cyan]pip install parallel-web-tools[/cyan]")
+            console.print("Also requires: gcloud CLI installed and authenticated")
             raise click.Abort() from None
 
         console.print(f"[bold cyan]Deploying to BigQuery in {project}...[/bold cyan]\n")

From b0bcec7b011c35a23f303fce29b0bec3853809a8 Mon Sep 17 00:00:00 2001
From: Matt Harris <mharris@parallel.ai>
Date: Sat, 24 Jan 2026 21:58:26 -0500
Subject: [PATCH 11/13] fix: limit source types in standalone CLI to csv and
 duckdb

BigQuery source type requires sqlalchemy-bigquery driver which isn't
bundled in standalone CLI. Conditionally show only available source
types based on sys.frozen check.
---
 parallel-web-tools.spec            |  2 +-
 parallel_web_tools/cli/commands.py | 11 +++++++++--
 2 files changed, 10 insertions(+), 3 deletions(-)

diff --git a/parallel-web-tools.spec b/parallel-web-tools.spec
index 756361c..fdb421c 100644
--- a/parallel-web-tools.spec
+++ b/parallel-web-tools.spec
@@ -37,10 +37,10 @@ a = Analysis(
         'parallel_web_tools.cli.commands',
         'parallel_web_tools.cli.planner',
         # Processors (for local file/db enrichment)
+        # Note: bigquery processor not included - requires sqlalchemy-bigquery driver
         'parallel_web_tools.processors',
         'parallel_web_tools.processors.csv',
         'parallel_web_tools.processors.duckdb',
-        'parallel_web_tools.processors.bigquery',
         # Note: Deploy commands (bigquery, snowflake) are NOT included in standalone CLI
         # They require: pip install parallel-web-tools[snowflake] or [bigquery]
         # Dependencies that might not be auto-detected
diff --git a/parallel_web_tools/cli/commands.py b/parallel_web_tools/cli/commands.py
index 6ba67af..1e6dc9f 100644
--- a/parallel_web_tools/cli/commands.py
+++ b/parallel_web_tools/cli/commands.py
@@ -36,6 +36,13 @@
 
 load_dotenv(".env.local")
 
+# Source types available for enrich run/plan
+# BigQuery requires sqlalchemy-bigquery driver which isn't in standalone CLI
+if getattr(sys, "frozen", False):
+    AVAILABLE_SOURCE_TYPES = ["csv", "duckdb"]
+else:
+    AVAILABLE_SOURCE_TYPES = ["csv", "duckdb", "bigquery"]
+
 
 # =============================================================================
 # Output Helpers
@@ -507,7 +514,7 @@ def enrich():
 
 @enrich.command(name="run")
 @click.argument("config_file", required=False)
-@click.option("--source-type", type=click.Choice(["csv", "duckdb", "bigquery"]), help="Data source type")
+@click.option("--source-type", type=click.Choice(AVAILABLE_SOURCE_TYPES), help="Data source type")
 @click.option("--source", help="Source file path or table name")
 @click.option("--target", help="Target file path or table name")
 @click.option("--source-columns", help="Source columns as JSON")
@@ -629,7 +636,7 @@ def enrich_run(
 
 @enrich.command(name="plan")
 @click.option("-o", "--output", default="config.yaml", help="Output YAML file path", show_default=True)
-@click.option("--source-type", type=click.Choice(["csv", "duckdb", "bigquery"]), help="Data source type")
+@click.option("--source-type", type=click.Choice(AVAILABLE_SOURCE_TYPES), help="Data source type")
 @click.option("--source", help="Source file path or table name")
 @click.option("--target", help="Target file path or table name")
 @click.option("--source-columns", help="Source columns as JSON")

From bdbe7e627f31d891dbf523953cb040c34c52fcdb Mon Sep 17 00:00:00 2001
From: Matt Harris <mharris@parallel.ai>
Date: Sat, 24 Jan 2026 22:10:50 -0500
Subject: [PATCH 12/13] fix: graceful error when CLI deps not installed

Base pip install is now library-only. Running parallel-cli without
the [cli] extra shows helpful message instead of ImportError.
---
 parallel_web_tools/cli/__init__.py | 13 ++++++++++++-
 1 file changed, 12 insertions(+), 1 deletion(-)

diff --git a/parallel_web_tools/cli/__init__.py b/parallel_web_tools/cli/__init__.py
index 80141f3..3ed78f1 100644
--- a/parallel_web_tools/cli/__init__.py
+++ b/parallel_web_tools/cli/__init__.py
@@ -1,5 +1,16 @@
 """CLI for Parallel Data."""
 
-from parallel_web_tools.cli.commands import main
+try:
+    from parallel_web_tools.cli.commands import main
+except ImportError:
+    import sys
+
+    def main():
+        """Stub for when CLI dependencies are not installed."""
+        print("parallel-cli requires additional dependencies.")
+        print("")
+        print("Install with: pip install parallel-web-tools[cli]")
+        sys.exit(1)
+
 
 __all__ = ["main"]

From bf32a58acef1e5e6438bc1cb3adb78b2568619e1 Mon Sep 17 00:00:00 2001
From: Matt Harris <mharris@parallel.ai>
Date: Sat, 24 Jan 2026 22:14:57 -0500
Subject: [PATCH 13/13] refactor: move CLI deps to base dependencies

CLI (click, rich, questionary, httpx) now included in base install.
Simpler user experience - parallel-cli always works after pip install.
Keep [cli] extra for backwards compatibility (empty).
---
 parallel_web_tools/cli/__init__.py            | 13 +---------
 .../bigquery/cloud_function/requirements.txt  |  2 +-
 pyproject.toml                                | 15 +++++------
 uv.lock                                       | 26 +++++++------------
 4 files changed, 18 insertions(+), 38 deletions(-)

diff --git a/parallel_web_tools/cli/__init__.py b/parallel_web_tools/cli/__init__.py
index 3ed78f1..80141f3 100644
--- a/parallel_web_tools/cli/__init__.py
+++ b/parallel_web_tools/cli/__init__.py
@@ -1,16 +1,5 @@
 """CLI for Parallel Data."""
 
-try:
-    from parallel_web_tools.cli.commands import main
-except ImportError:
-    import sys
-
-    def main():
-        """Stub for when CLI dependencies are not installed."""
-        print("parallel-cli requires additional dependencies.")
-        print("")
-        print("Install with: pip install parallel-web-tools[cli]")
-        sys.exit(1)
-
+from parallel_web_tools.cli.commands import main
 
 __all__ = ["main"]
diff --git a/parallel_web_tools/integrations/bigquery/cloud_function/requirements.txt b/parallel_web_tools/integrations/bigquery/cloud_function/requirements.txt
index e9ae780..fe3b239 100644
--- a/parallel_web_tools/integrations/bigquery/cloud_function/requirements.txt
+++ b/parallel_web_tools/integrations/bigquery/cloud_function/requirements.txt
@@ -1,5 +1,5 @@
 # Cloud Function dependencies for BigQuery Remote Function
 functions-framework>=3.0.0
 flask>=3.0.0
-parallel-web-tools>=0.0.5
+parallel-web-tools>=0.0.4
 google-cloud-secret-manager>=2.20.0
diff --git a/pyproject.toml b/pyproject.toml
index f6bbecf..456e4f4 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -32,6 +32,11 @@ dependencies = [
     "pyarrow>=18.0.0",
     "python-dotenv>=1.0.0",
     "pyyaml>=6.0.0",
+    # CLI dependencies
+    "click>=8.1.0",
+    "questionary>=2.0.0",
+    "rich>=13.0.0",
+    "httpx>=0.25.0",
 ]
 
 [project.urls]
@@ -44,13 +49,6 @@ Issues = "https://github.com/parallel-web/parallel-web-tools/issues"
 parallel-cli = "parallel_web_tools.cli:main"
 
 [project.optional-dependencies]
-# CLI extras (interactive features)
-cli = [
-    "click>=8.1.0",
-    "questionary>=2.0.0",
-    "rich>=13.0.0",
-    "pyrefly>=0.49.0",
-]
 # Polars integration (already in main deps, this is for explicit install)
 polars = []  # polars>=1.37.0 is in main dependencies
 # DuckDB integration
@@ -74,7 +72,7 @@ spark = [
 ]
 # All features (default for CLI install)
 all = [
-    "parallel-web-tools[cli,polars,duckdb,snowflake,bigquery]",
+    "parallel-web-tools[polars,duckdb,snowflake,bigquery]",
 ]
 # Development
 dev = [
@@ -84,6 +82,7 @@ dev = [
     "pyinstaller>=6.0.0",
     "pre-commit>=4.0.0",
     "ruff>=0.14.0",
+    "pyrefly>=0.49.0",
 ]
 
 [tool.hatch.build.targets.wheel]
diff --git a/uv.lock b/uv.lock
index 903b1f9..21e75d3 100644
--- a/uv.lock
+++ b/uv.lock
@@ -1060,21 +1060,21 @@ name = "parallel-web-tools"
 version = "0.0.5"
 source = { editable = "." }
 dependencies = [
+    { name = "click" },
+    { name = "httpx" },
     { name = "pandas" },
     { name = "parallel-web" },
     { name = "polars" },
     { name = "pyarrow" },
     { name = "python-dotenv" },
     { name = "pyyaml" },
+    { name = "questionary" },
+    { name = "rich" },
 ]
 
 [package.optional-dependencies]
 all = [
-    { name = "click" },
     { name = "duckdb" },
-    { name = "pyrefly" },
-    { name = "questionary" },
-    { name = "rich" },
     { name = "snowflake-connector-python" },
     { name = "sqlalchemy" },
     { name = "sqlalchemy-bigquery" },
@@ -1083,14 +1083,7 @@ bigquery = [
     { name = "sqlalchemy" },
     { name = "sqlalchemy-bigquery" },
 ]
-cli = [
-    { name = "click" },
-    { name = "pyrefly" },
-    { name = "questionary" },
-    { name = "rich" },
-]
 dev = [
-    { name = "click" },
     { name = "duckdb" },
     { name = "httpx" },
     { name = "pre-commit" },
@@ -1099,8 +1092,6 @@ dev = [
     { name = "pyspark" },
     { name = "pytest" },
     { name = "pytest-cov" },
-    { name = "questionary" },
-    { name = "rich" },
     { name = "ruff" },
     { name = "snowflake-connector-python" },
     { name = "sqlalchemy" },
@@ -1124,8 +1115,9 @@ dev = [
 
 [package.metadata]
 requires-dist = [
-    { name = "click", marker = "extra == 'cli'", specifier = ">=8.1.0" },
+    { name = "click", specifier = ">=8.1.0" },
     { name = "duckdb", marker = "extra == 'duckdb'", specifier = ">=1.0.0" },
+    { name = "httpx", specifier = ">=0.25.0" },
     { name = "httpx", marker = "extra == 'spark'", specifier = ">=0.25.0" },
     { name = "pandas", specifier = ">=2.3.0" },
     { name = "parallel-web", specifier = ">=0.4.0" },
@@ -1135,14 +1127,14 @@ requires-dist = [
     { name = "pre-commit", marker = "extra == 'dev'", specifier = ">=4.0.0" },
     { name = "pyarrow", specifier = ">=18.0.0" },
     { name = "pyinstaller", marker = "extra == 'dev'", specifier = ">=6.0.0" },
-    { name = "pyrefly", marker = "extra == 'cli'", specifier = ">=0.49.0" },
+    { name = "pyrefly", marker = "extra == 'dev'", specifier = ">=0.49.0" },
     { name = "pyspark", marker = "extra == 'spark'", specifier = ">=3.4.0" },
     { name = "pytest", marker = "extra == 'dev'", specifier = ">=8.0.0" },
     { name = "pytest-cov", marker = "extra == 'dev'", specifier = ">=4.0.0" },
     { name = "python-dotenv", specifier = ">=1.0.0" },
     { name = "pyyaml", specifier = ">=6.0.0" },
-    { name = "questionary", marker = "extra == 'cli'", specifier = ">=2.0.0" },
-    { name = "rich", marker = "extra == 'cli'", specifier = ">=13.0.0" },
+    { name = "questionary", specifier = ">=2.0.0" },
+    { name = "rich", specifier = ">=13.0.0" },
     { name = "ruff", marker = "extra == 'dev'", specifier = ">=0.14.0" },
     { name = "snowflake-connector-python", marker = "extra == 'snowflake'", specifier = ">=3.0.0" },
     { name = "sqlalchemy", marker = "extra == 'bigquery'", specifier = ">=2.0.0" },