How to Ingest CSV into CrewAI with EnConvert

import os
import json
import re
import time

import requests
from crewai import Agent, Crew, Task
from crewai.tools import BaseTool
from pydantic import BaseModel, Field

HEADERS = {"X-API-Key": os.environ["ENCONVERT_API_KEY"]}

# Step 1: Upload the CSV
# The multipart field is "files". The CSV becomes one Markdown table that stays
# whole as a chunk (tables are never split). Tables are capped at 5000 data rows.
with open("/path/to/your/crm-export.csv", "rb") as f:
    upload = requests.post(
        "https://api.enconvert.com/v2/ingest/files",
        headers=HEADERS,
        files={"files": ("crm-export.csv", f, "text/csv")},
        timeout=60,
    )
upload.raise_for_status()
job_id = upload.json()["job_id"]

# Step 2: Poll until the job completes (give up after 5 minutes)
deadline = time.monotonic() + 300
while True:
    status_response = requests.get(
        f"https://api.enconvert.com/v2/ingest/{job_id}",
        headers=HEADERS,
        timeout=60,
    )
    status_response.raise_for_status()
    job = status_response.json()
    if job["status"] == "completed":
        output_url = job["output_url"]
        break
    if job["status"] in ("failed", "canceled"):
        raise RuntimeError(job.get("error_message") or f"Ingest job {job['status']}")
    if time.monotonic() > deadline:
        raise RuntimeError(f"Ingest job {job_id} still {job['status']} after 5 minutes")
    time.sleep(2)

# Step 3: Download the JSONL output
# The output_url is a signed URL (valid 15 minutes). Plain GET, no API key.
jsonl = requests.get(output_url, timeout=60)
jsonl.raise_for_status()

chunks = [
    json.loads(line)
    for line in jsonl.content.decode("utf-8").split("\n")
    if line.strip()
]
print(f"Loaded {len(chunks)} chunks ready for retrieval")


# Step 4: Define the input schema and BaseTool
def count_matches(text: str, terms: set[str]) -> int:
    return len(terms & set(re.findall(r"\w+", text.casefold())))


class SearchInput(BaseModel):
    query: str = Field(
        description="What to look up in the CSV, such as a customer name or a status value"
    )
    limit: int = Field(
        default=10,
        ge=1,
        le=50,
        description="Maximum number of table rows to return",
    )


class SearchCSVTool(BaseTool):
    name: str = "search_csv"
    description: str = (
        "Search the ingested CSV and return the best matching rows as a Markdown "
        "table, always including the column header row."
    )
    args_schema: type[BaseModel] = SearchInput

    def _run(self, query: str, limit: int = 10) -> str:
        terms = set(re.findall(r"\w+", query.casefold()))
        results = []

        for chunk in chunks:
            # A table chunk is the header row, the separator row, then the data rows.
            # Skip any other chunk, such as the note added when the table is truncated.
            lines = chunk["content"].split("\n")
            if len(lines) < 2 or not lines[0].lstrip().startswith("|"):
                continue
            header, separator, *rows = lines
            matches = [row for row in rows if count_matches(row, terms)]
            matches.sort(key=lambda row: count_matches(row, terms), reverse=True)

            if matches:
                table = "\n".join([header, separator, *matches[:limit]])
                # For a file, title is the filename. headings_path is [] for a CSV
                # (one table, no headings); an XLSX adds one "## SheetName" section
                # per sheet, which shows up here as "pricing.xlsx > Sheet1".
                metadata = chunk["metadata"]
                label = " > ".join([metadata["title"], *metadata["headings_path"]])
                results.append(f"[Source: {label}]\n{table}")

        return "\n\n---\n\n".join(results) if results else "No matching CSV rows found."


# Replace keyword matching with embedding search for larger files

# Step 5: Set up the agent
# CrewAI uses OpenAI by default and reads OPENAI_API_KEY from the environment
analyst = Agent(
    role="CRM Data Analyst",
    goal="Answer questions about the CRM export accurately, using the column headers in each table",
    backstory="You read tabular exports and report values together with their column names.",
    tools=[SearchCSVTool()],
    verbose=True,
)

# Step 6: Create a task and run the crew
# Replace the question with one about the columns in your own CSV
task = Task(
    description="List the accounts with an open renewal status and give the status of each.",
    expected_output="A short list of accounts with their renewal status.",
    agent=analyst,
)

crew = Crew(agents=[analyst], tasks=[task], verbose=True)
print(crew.kickoff())