How to Integrate AI Agents with Airtable

pip install pyairtable langchain langchain-openai python-dotenv

export AIRTABLE_API_KEY="patXXXXXXXXXXXXXX.your_token"
export AIRTABLE_BASE_ID="appXXXXXXXXXXXXXX"

from pyairtable import Api

api = Api(os.getenv("AIRTABLE_API_KEY"))
base = api.base(os.getenv("AIRTABLE_BASE_ID"))
table = base.table("Contacts")
print(table.first())  # Should print the first record

import os
from pyairtable import Api
from langchain.tools import tool
from dotenv import load_dotenv

load_dotenv()

api = Api(os.getenv("AIRTABLE_API_KEY"))
base = api.base(os.getenv("AIRTABLE_BASE_ID"))


@tool
def search_records(table_name: str, search_field: str, search_value: str) -> str:
    """
    Search Airtable records in a table by field value.
    Returns matching records with their IDs and field values.
    """
    table = base.table(table_name)
    formula = f"SEARCH(LOWER('{search_value}'), LOWER({{{search_field}}}))"
    records = table.all(formula=formula, max_records=10)

    if not records:
        return f"No records found in {table_name} where {search_field} contains '{search_value}'"

    result = [f"Found {len(records)} records in {table_name}:"]
    for rec in records:
        fields = rec["fields"]
        field_summary = " | ".join(f"{k}: {str(v)[:50]}" for k, v in list(fields.items())[:5])
        result.append(f"  ID: {rec['id']} | {field_summary}")
    return "\n".join(result)


@tool
def create_record(table_name: str, fields: dict) -> str:
    """
    Create a new record in an Airtable table.
    fields is a dict mapping field names to values.
    """
    table = base.table(table_name)
    record = table.create(fields)
    created_fields = record.get("fields", {})
    primary_field = list(created_fields.values())[0] if created_fields else "Unknown"
    return f"Record created in {table_name}: '{primary_field}' (ID: {record['id']})"


@tool
def update_record(table_name: str, record_id: str, fields: dict) -> str:
    """Update specific fields on an existing Airtable record by ID."""
    table = base.table(table_name)
    updated = table.update(record_id, fields)
    return f"Record {record_id} updated in {table_name}. Updated fields: {list(fields.keys())}"


@tool
def get_records_by_view(table_name: str, view_name: str, max_records: int = 20) -> str:
    """
    Fetch records from a specific Airtable view (e.g., 'Active Leads', 'This Week').
    Views pre-filter and sort records based on configured criteria.
    """
    table = base.table(table_name)
    records = table.all(view=view_name, max_records=max_records)

    if not records:
        return f"No records in view '{view_name}' of table {table_name}"

    result = [f"Records from {table_name} / {view_name} ({len(records)} shown):"]
    for rec in records[:15]:
        fields = rec["fields"]
        field_str = " | ".join(f"{k}: {str(v)[:40]}" for k, v in list(fields.items())[:4])
        result.append(f"  {field_str}")
    return "\n".join(result)


@tool
def batch_create_records(table_name: str, records_data: list) -> str:
    """
    Create multiple records at once in Airtable. records_data is a list of field dicts.
    More efficient than creating records one-by-one.
    """
    table = base.table(table_name)
    # Airtable batch allows up to 10 records per call
    created_count = 0
    for i in range(0, len(records_data), 10):
        batch = records_data[i:i+10]
        table.batch_create(batch)
        created_count += len(batch)
    return f"Created {created_count} records in {table_name}"


@tool
def find_duplicate_records(table_name: str, dedup_field: str) -> str:
    """Find potential duplicate records based on a field value match."""
    table = base.table(table_name)
    all_records = table.all(fields=[dedup_field])

    value_counts: dict = {}
    for rec in all_records:
        val = str(rec["fields"].get(dedup_field, "")).strip().lower()
        if val:
            value_counts.setdefault(val, []).append(rec["id"])

    duplicates = {v: ids for v, ids in value_counts.items() if len(ids) > 1}
    if not duplicates:
        return f"No duplicate {dedup_field} values found in {table_name}"

    lines = [f"Duplicate {dedup_field} values in {table_name}:"]
    for val, ids in list(duplicates.items())[:10]:
        lines.append(f"  '{val}': {len(ids)} records ({', '.join(ids[:3])})")
    return "\n".join(lines)

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o", temperature=0.1)
tools = [search_records, create_record, update_record,
         get_records_by_view, batch_create_records, find_duplicate_records]

prompt = ChatPromptTemplate.from_messages([
    ("system", """You are a data operations specialist with access to an Airtable database.

When managing records:
- Search before creating to avoid duplicates
- Structure data consistently using existing field names
- Batch create when processing multiple records for efficiency
- Flag data quality issues rather than silently skipping records

When reporting:
- Use view-based queries to access pre-filtered datasets
- Summarize record counts and field value distributions
- Identify patterns and anomalies in the data"""),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True, max_iterations=6)

# Example: Process a batch of leads
result = executor.invoke({
    "input": """Process these leads from today's webinar sign-ups and create records in the Leads table:
1. John Smith, Acme Corp, CTO, interested in enterprise plan
2. Sarah Lee, StartupXYZ, Head of Product, free trial request
3. Mike Johnson, TechCo, Developer, looking for API access

Search for existing records first to avoid duplicates."""
})
print(result["output"])