How to Summarise Twilio Docs with CrewAI

import os

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

BASE_URL = "https://api.enconvert.com/v2"
HEADERS = {
    "X-API-Key": os.environ["ENCONVERT_API_KEY"],
    "Content-Type": "application/json",
}

DOC_SCHEMA = {
    "title": "The page title",
    "body": "The main descriptive text of the page, without navigation or sidebar content",
    "endpoint_or_method": "The HTTP method and path of the API resource, for example POST /2010-04-01/Accounts/{AccountSid}/Messages.json",
    "parameters": "The request parameters listed on the page, with their types and whether they are required",
    "response_fields": "The properties returned in the resource, with their types",
}

class TwilioDocsInput(BaseModel):
    urls: list[str] = Field(description="Twilio documentation page URLs to read, at most 50")

class TwilioDocsTool(BaseTool):
    name: str = "read_twilio_docs"
    description: str = (
        "Reads Twilio API documentation pages and returns structured fields for each page: "
        "title, body, endpoint or method, parameters, and response fields."
    )
    args_schema: type[BaseModel] = TwilioDocsInput

    def _run(self, urls: list[str]) -> str:
        if len(urls) > 50:
            return "Too many URLs. Distill accepts at most 50 per call."

        # One synchronous call: no job_id, no polling, no download step
        response = requests.post(
            f"{BASE_URL}/distill",
            headers=HEADERS,
            json={"urls": urls, "schema": DOC_SCHEMA},
            timeout=120,  # increase for larger batches
        )

        # Distill is available on paid plans only, and the free tier returns 402
        if response.status_code == 402:
            return "Distill requires a paid plan."
        response.raise_for_status()

        pages = []
        for result in response.json()["results"]:
            # Failed pages come back as rows with a failed status, not as errors
            if result["status"] != "completed":
                pages.append(f"Source: {result['url']}\nCould not be read ({result['status']}).")
                continue

            rq = result.get("render_quality", 1.0)
            if rq < 0.5:
                # a low score can mean a login wall or challenge page
                pages.append(f"Source: {result['url']}\nLow render quality ({rq:.2f}), skipped.")
                continue

            data = result.get("data") or {}  # keys match schema exactly: title, body, endpoint_or_method, parameters, response_fields
            pages.append(
                f"Source: {result['url']}\n"
                f"Title: {data.get('title')}\n"
                f"Endpoint: {data.get('endpoint_or_method')}\n"
                f"Parameters: {data.get('parameters')}\n"
                f"Response fields: {data.get('response_fields')}\n\n"
                f"{data.get('body')}"
            )

        if not pages:
            return "No Twilio documentation pages could be extracted."

        return "\n\n---\n\n".join(pages)


# CrewAI's default LLM reads OPENAI_API_KEY from the environment
docs_reader = Agent(
    role="Twilio API Analyst",
    goal="Answer questions about the Twilio API accurately, using only the documentation pages you read",
    backstory="You read API reference pages closely and cite the page behind every statement.",
    tools=[TwilioDocsTool()],
    allow_delegation=False,
    verbose=True,
    llm="gpt-4.1",
)

task = Task(
    description=(
        "Read these pages: https://www.twilio.com/docs/messaging/api/message-resource and "
        "https://www.twilio.com/docs/voice/api/call-resource. "
        "Compare the required parameters for sending a message and for creating a call, "
        "and cite the page for each parameter."
    ),
    expected_output="A short comparison of the required parameters for each resource, with sources.",
    agent=docs_reader,
)

crew = Crew(
    agents=[docs_reader],
    tasks=[task],
    process=Process.sequential,
    verbose=True,
)
print(crew.kickoff())