Building an AI-Powered Lead Management System with External API Integration

Abstract

Modern applications rarely operate in isolation. Instead of building every function from scratch, developers often rely on external APIs to add powerful capabilities, such as accessing AI models, storing data in cloud services, or integrating with third-party platforms. Calling external APIs is a necessary skill in backend development: it allows your system to connect with specialized services and extend functionality far beyond what your own server provides.

In this article, we demonstrate how to design an AI-powered assistant that depends on two such external APIs:

• The OpenAI API, which interprets unstructured user conversations and generates both natural-language replies and structured data.
• The Airtable API, which stores that structured data as persistent lead records.

In this article, we will create a conversation thread, capture user input, generate responses with AI, save leads into Airtable, handle API errors, and show how each step is implemented in Python code.

Introduction

When building conversational assistants, we need more than just natural language responses and we often want structured data. For example, a user might tell the chatbot: “I’m in Shanghai, I want to study AI courses, and my budget is 6000.”

From this, the assistant should:

1. Reply naturally in chat.
2. Extract structured fields (name, phone, summary, intention).
3. Save them into Airtable as a lead record.

We achieve this by combining:

• OpenAI API (for reasoning and structured output).
• Airtable API (for persistence).
• Python backend (FastAPI, Flask, etc. to orchestrate).

Workflow

Let’s map the assistant’s lifecycle into five parts:

Create Thread:

Before the assistant can engage with a user, it must create a conversation thread. This is like opening a new chat room with a unique ID, where all future messages will be grouped together. Without threads, multiple conversations would mix together, making it impossible to track context.

Capture User Input:

Once the thread is initialized, the assistant’s next task is to capture what the user says. This step is about storing the user’s raw input into the thread’s history. Capturing input ensures the assistant has memory of the conversation. When the AI generates a response, it can use not only the latest message but also prior ones.

Generate Response:

This is the core of the system. The assistant sends the captured user input to the OpenAI API, which generates both:

• A natural-language reply (reply) → what the chatbot says back to the user.
• A structured data object (lead) → fields like name, phone, budget, or intention, extracted from the user’s message.

Natural conversation alone is not enough for business workflows. By combining chat replies with structured outputs, the system can act on data (like storing a new lead).

Send Response:

Once the AI generates a response, the system needs to send it back to the user. This is where the assistant closes the loop for each message. The user expects immediate feedback. A chatbot that silently processes data but doesn’t reply will feel broken.

API Request Failed:

Not all requests succeed. An Airtable API call might fail due to invalid credentials, or OpenAI might reject a request due to rate limits. The system must gracefully handle these cases. Without error handling, users would see silence or crashes, breaking trust in the assistant.

Implementation in Code

• Create Thread (GET Request)： We generate a unique thread_id so each user session has its own context.

from fastapi import FastAPI, HTTPException
from uuid import uuid4

app = FastAPI()
THREAD_STORE = {}

@app.get("/thread")
def create_thread():
    try:
        thread_id = str(uuid4())
        THREAD_STORE[thread_id] = {"messages": []}
        return {"thread_id": thread_id}
    except Exception:
        raise HTTPException(status_code=500, detail="Failed to create new thread")

• Capture User Input (POST Request): The frontend sends the thread_id and user_message.

from pydantic import BaseModel

class ChatRequest(BaseModel):
    thread_id: str
    user_message: str

@app.post("/chat")
def chat(req: ChatRequest):
    if req.thread_id not in THREAD_STORE:
        raise HTTPException(status_code=404, detail="Thread not found")

    THREAD_STORE[req.thread_id]["messages"].append(
        {"role": "user", "content": req.user_message}
    )

    try:
        assistant_text, extracted = call_llm_and_maybe_create_lead(
            req.user_message, req.thread_id
        )
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"AI generation failed: {e}")

    THREAD_STORE[req.thread_id]["messages"].append(
        {"role": "assistant", "content": assistant_text}
    )
    return {
        "thread_id": req.thread_id,
        "assistant_reply": assistant_text,
        "extracted_fields": extracted
    }

• Generate Response (OpenAI API)

We instruct the AI to output JSON with two keys:

• "reply" → plain text to send to the user.
• "lead" → structured lead data.

import os, json, time
from openai import OpenAI
from prompts import assistant_instructions

client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

def call_llm_and_maybe_create_lead(user_message: str, thread_id: str):
    system_msg = {
        "role": "system",
        "content": f"""{assistant_instructions}
Output valid JSON with keys: reply, lead."""
    }
    user_msg = {"role": "user", "content": user_message}

    body = {
        "model": "gpt-4o-mini",
        "messages": [system_msg, user_msg],
        "temperature": 0.2
    }

    resp = client.chat.completions.create(**body)
    content = resp.choices[0].message.content
    data = json.loads(content)

    assistant_reply = data.get("reply", "")
    extracted = data.get("lead", {})

    maybe_write_airtable(extracted)
    return assistant_reply, extracted

• Save Lead to Airtable: If required fields (summary, intention) are present, we insert into Airtable.

import requests

def create_lead(name, phone, wechat, address, summary, intention):
    url = os.environ["AIRTABLE_URL"]
    headers = {
        "Authorization": os.environ["AIRTABLE_API_KEY"],
        "Content-Type": "application/json"
    }
    data = {
        "records": [{
            "fields": {
                "Name": name,
                "Phone": phone,
                "WeChat": wechat,
                "Address": address,
                "Summary": summary,
                "Intention": intention
            }
        }]
    }
    response = requests.post(url, headers=headers, json=data)
    return response

def maybe_write_airtable(extracted: dict):
    if not extracted:
        return
    payload = {
        "name": extracted.get("name", ""),
        "phone": extracted.get("phone", ""),
        "wechat": extracted.get("wechat", ""),
        "address": extracted.get("address", ""),
        "summary": extracted.get("summary", ""),
        "intention": extracted.get("intention", "")
    }
    if not payload["summary"] or not payload["intention"]:
        return
    resp = create_lead(**payload)
    if resp.status_code == 200:
        print("Lead created successfully.")
    else:
        print(f"Failed to create lead: {resp.text}")

• Error Handling (API Request Failed)
• If /thread fails → return HTTP 500.
• If /chat fails → return HTTP 502.
• If Airtable fails → log and skip record creation.

Conclusion

External APIs like OpenAI and Airtable behave much like your own backend APIs — they’re just HTTP endpoints. By designing a workflow (thread creation, input capture, response generation, persistence, and error handling) and mapping it to Python code, we built a system that automates lead management end-to-end.

The **reply field provides a user-facing answer, while the `lead`** object ensures structured data entry. This dual-output approach enables chatbots not only to “talk” but also to “work” in real business pipelines.