How to Scrape Salesforce with Claude Code

import os

import anthropic
import requests

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

SALESFORCE_URLS = [url.strip() for url in os.environ["SALESFORCE_PAGE_URLS"].split(",")]  # public Salesforce documentation or Knowledge article URLs

client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

tools = [
    {
        "name": "read_salesforce_pages",
        "description": (
            "Reads publicly accessible Salesforce documentation or Knowledge article pages "
            "and returns the main content of each as clean Markdown, labelled with its source URL."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "urls": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "Salesforce page URLs to read, up to 10 per call",
                }
            },
            "required": ["urls"],
        },
    }
]

def read_salesforce_pages(urls: list[str]) -> str:
    if len(urls) > 10:
        raise ValueError("This tool covers inline batches of 10 URLs or fewer")

    response = requests.post(
        f"{BASE_URL}/perceive/batch",
        headers=enconvert_headers,
        json={
            "urls": urls,
            "options": {"outputs": ["markdown"]},
        },
        timeout=300,
    )

    if response.status_code == 402:
        return "Monthly ops allowance exhausted"
    response.raise_for_status()

    sections = []
    for result in response.json()["items"]:
        page_url = result["url"]

        if result.get("status") != "completed":
            sections.append(f"Source: {page_url}\nRender failed ({result.get('status')}).")
            continue

        if result.get("is_blocked"):
            sections.append(f"Source: {page_url}\nBlocked page. No content returned.")
            continue

        render_quality = result.get("render_quality", 1.0)
        if render_quality < 0.40:
            sections.append(f"Source: {page_url}\nRender failed (quality {render_quality:.2f}).")
            continue

        # Markdown is a pre-signed URL; fetch the content in a second request
        markdown_url = result["outputs"]["markdown"]["url"]
        markdown = requests.get(markdown_url, timeout=60)
        markdown.raise_for_status()

        sections.append(f"Source: {page_url}\n\n{markdown.text}")

    return "\n\n---\n\n".join(sections) or "No pages were returned."

messages = [
    {
        "role": "user",
        "content": (
            f"Read these Salesforce pages: {', '.join(SALESFORCE_URLS)}. "
            "Explain the main configuration steps, citing each page."
        ),
    }
]

# Agentic loop: call the model, run any tool it requests, send the result back
while True:
    response = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=4096,
        system="Answer only from the pages you read, and cite the source URL for every statement.",
        tools=tools,
        messages=messages,
    )

    tool_results = []
    for block in response.content:
        if block.type == "tool_use":
            tool_results.append(
                {
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": read_salesforce_pages(**block.input),
                }
            )

    if not tool_results:
        break

    messages.append({"role": "assistant", "content": response.content})
    messages.append({"role": "user", "content": tool_results})

print("".join(block.text for block in response.content if block.type == "text"))