Skip to content
Home
Generative AI
Python

Beyond Fixed Schemas: Architecting a Self‑Correcting Agentic Pipeline for Evolving GitHub APIs

Mausam

When a third‑party API adds a field, renames an attribute, or throttles you for a moment, your nightly ETL job either crashes or silently drops data. You’ve probably patched the script dozens of times, only to discover the next change breaks it again. I faced that exact headache while pulling repository metrics from GitHub for a dashboard that feeds into our forecasting models. In this guide I show you how to let Claude act as a resilient middle‑man: it calls a robust wrapper around the GitHub Repo API, extracts the fields you care about even when the JSON shape shifts, validates the result against a strict Pydantic model, and feeds any validation errors back to Claude for a second pass. The result is a production‑ready extractor that fixes itself without a line of manual code.

Key Takeaways

  • Encapsulate HTTP calls with exponential backoff and rate‑limit handling so the agent never sees a raw network error.
  • Guide Claude with a system prompt that anticipates schema variations and forces a canonical JSON output.
  • Use Pydantic validation as the single source of truth; feed validation errors back to Claude to trigger self‑correction.
  • Wrap the whole flow in a loop that caps retries, logs each attempt, and surfaces a clean, typed object to downstream pipelines.
  • The pattern scales to any external API—just swap the tool and adjust the schema.

The Problem

Our data platform ingests GitHub repository statistics every hour to enrich a financial model that predicts open‑source contributions to the local tech ecosystem. The upstream API occasionally adds new keys (e.g., archived) or deprecates old ones (e.g., watchers_count → subscribers_count). A static Python script that expects stargazers_count and forks_count would raise a KeyError and halt the pipeline, causing stale data downstream. We needed a solution that could:

  1. Make the HTTP request reliably, respecting rate limits.
  2. Extract the required fields even if their names change.
  3. Validate the extracted payload against a contract we control.
  4. Iterate automatically until the contract is satisfied.

Data and Sources

We pull live data from the public GitHub Repository API:
https://api.github.com/repos/python/cpython
Documentation: GitHub REST API – Get a repository.
Data accessed on 2026-10-22.

Loading the Data

The first building block is a resilient HTTP helper that Claude can call as a tool. It retries on transient failures, backs off when GitHub returns 403 for rate limiting, and always returns a dictionary – either the raw JSON or a structured error.

import requests
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

class APIError(RuntimeError):
    pass

@retry(
    reraise=True,
    stop=stop_after_attempt(4),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    retry=retry_if_exception_type(requests.exceptions.RequestException),
)
def get_github_repo_data(owner: str, repo: str) -> dict:
    url = f"https://api.github.com/repos/{owner}/{repo}"
    try:
        resp = requests.get(url, timeout=5)
        if resp.status_code == 403 and "rate limit" in resp.text.lower():
            raise APIError("Rate limit exceeded")
        resp.raise_for_status()
        return resp.json()
    except requests.exceptions.RequestException as exc:
        raise APIError(f"Network problem: {exc}") from exc

Claude sees this function as a @tool and can invoke it with any owner/repo pair. The decorator from tenacity guarantees exponential backoff, which is essential for production stability.

The Core Logic

Step 1 – Defining the Robust API Interaction Tool

This step isolates network unreliability from the LLM. By returning a plain dict, Claude can focus on semantic extraction without worrying about HTTP details.

Step 2 – Crafting the Adaptive Extraction Agent Prompt

We give Claude a system prompt that tells it exactly what to do: call the tool, look for several possible field names, and always output a strict JSON payload. The prompt also asks Claude to flag missing critical fields.

system_prompt = """
You are an extraction assistant for the GitHub Repo API.
Your goal is to produce a JSON object matching this schema:
{
  "name": string,
  "description": string | null,
  "stars": integer,
  "forks": integer,
  "open_issues": integer,
  "last_updated_at": string (ISO 8601)
}
When you call the tool `get_github_repo_data`, you will receive the raw JSON.
Map the following possible keys to the schema:
  - name -> name
  - description -> description
  - stargazers_count OR stars OR star_count -> stars
  - forks_count OR forks -> forks
  - open_issues_count OR open_issues OR issues -> open_issues
  - updated_at OR pushed_at -> last_updated_at
If any required field is missing, include a key `"error": ""` in the output.
Never add extra keys.
Respond ONLY with the JSON object.
"""

The prompt is deliberately exhaustive about alternate key names because GitHub occasionally adds aliases. Claude’s “inner monologue” will be captured in the conversation log for debugging.

Step 3 – Implementing Agentic Data Standardization and Validation

After Claude returns a JSON string, we validate it with Pydantic. Validation errors are turned into a user message that Claude sees, prompting a second attempt.

from pydantic import BaseModel, Field, ValidationError
from datetime import datetime
from typing import Optional

class GitHubRepoData(BaseModel):
    name: str
    description: Optional[str] = None
    stars: int = Field(..., ge=0)
    forks: int = Field(..., ge=0)
    open_issues: int = Field(..., ge=0)
    last_updated_at: datetime

def validate_agent_output(json_str: str) -> GitHubRepoData:
    try:
        return GitHubRepoData.model_validate_json(json_str)
    except ValidationError as ve:
        # Return the error details as a string for Claude
        raise ValueError(str(ve))

The model is the single source of truth. If Claude mis‑spells a field or returns a string where an integer is expected, Pydantic raises a ValidationError that we capture and feed back.

Step 4 – Orchestrating the Self‑Correcting Pipeline Loop

The orchestration function ties everything together. It builds a Claude conversation, executes tool calls, validates the output, and retries up to a configurable limit.

import json
import logging
from anthropic import Anthropic, messages

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
client = Anthropic(api_key="YOUR_ANTHROPIC_API_KEY")

def run_adaptive_extraction_pipeline(owner: str, repo: str, max_retries: int = 3):
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": f"Extract data for {owner}/{repo}."}
    ]

    for attempt in range(1, max_retries + 1):
        response = client.messages.create(
            model="claude-3-5-sonnet-20240620",
            max_tokens=500,
            temperature=0,
            messages=messages,
        )
        tool_calls = response.content[0].get("tool_use")  # simplified extraction
        if tool_calls:
            # Execute the tool and append result
            tool_name = tool_calls["name"]
            args = tool_calls["input"]
            if tool_name == "get_github_repo_data":
                api_result = get_github_repo_data(args["owner"], args["repo"])
                messages.append({"role": "assistant", "content": json.dumps(api_result)})
                continue  # go back to Claude with raw data

        # Assume Claude's final reply is JSON
        agent_output = response.content[0].text.strip()
        try:
            validated = validate_agent_output(agent_output)
            logging.info("Validation succeeded on attempt %s", attempt)
            return validated
        except ValueError as ve:
            logging.warning("Validation failed: %s", ve)
            # Feed the error back and ask Claude to try again
            messages.append({
                "role": "user",
                "content": f"The previous JSON did not match the schema: {ve}. Please re‑format."
            })
    raise RuntimeError("Maximum retries exceeded")

The loop respects three failure modes:

  • Network/Rate‑limit errors: handled inside get_github_repo_data with retries.
  • Schema mismatches: caught by Pydantic and fed back to Claude.
  • Persistent LLM hallucination: after max_retries we abort, letting the ops team investigate.

Putting It Together

Running the pipeline is as simple as calling the orchestrator with a repository identifier. The function returns a typed GitHubRepoData instance that downstream code can safely consume.

if __name__ == "__main__":
    repo = run_adaptive_extraction_pipeline("python", "cpython")
    print(repo.json(indent=2))

Complete Script

The full runnable script combining all steps:

#!/usr/bin/env python3
"""
Self‑correcting GitHub Repo extractor using Claude Code Plugins.
Author: Mausam Adhikari
"""

import json
import logging
from datetime import datetime
from typing import Optional

import requests
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from pydantic import BaseModel, Field, ValidationError
from anthropic import Anthropic

# ----------------------------------------------------------------------
# Robust API tool
# ----------------------------------------------------------------------
class APIError(RuntimeError):
    """Raised when the GitHub API cannot be reached or returns a rate‑limit."""

@retry(
    reraise=True,
    stop=stop_after_attempt(4),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    retry=retry_if_exception_type(requests.exceptions.RequestException),
)
def get_github_repo_data(owner: str, repo: str) -> dict:
    url = f"https://api.github.com/repos/{owner}/{repo}"
    try:
        resp = requests.get(url, timeout=5)
        if resp.status_code == 403 and "rate limit" in resp.text.lower():
            raise APIError("Rate limit exceeded")
        resp.raise_for_status()
        return resp.json()
    except requests.exceptions.RequestException as exc:
        raise APIError(f"Network problem: {exc}") from exc

# ----------------------------------------------------------------------
# Pydantic schema
# ----------------------------------------------------------------------
class GitHubRepoData(BaseModel):
    name: str
    description: Optional[str] = None
    stars: int = Field(..., ge=0)
    forks: int = Field(..., ge=0)
    open_issues: int = Field(..., ge=0)
    last_updated_at: datetime

def validate_agent_output(json_str: str) -> GitHubRepoData:
    try:
        return GitHubRepoData.model_validate_json(json_str)
    except ValidationError as ve:
        raise ValueError(str(ve))

# ----------------------------------------------------------------------
# Claude system prompt
# ----------------------------------------------------------------------
system_prompt = """
You are an extraction assistant for the GitHub Repo API.
Your goal is to produce a JSON object matching this schema:
{
  "name": string,
  "description": string | null,
  "stars": integer,
  "forks": integer,
  "open_issues": integer,
  "last_updated_at": string (ISO 8601)
}
When you call the tool `get_github_repo_data`, you will receive the raw JSON.
Map the following possible keys to the schema:
  - name -> name
  - description -> description
  - stargazers_count OR stars OR star_count -> stars
  - forks_count OR forks -> forks
  - open_issues_count OR open_issues OR issues -> open_issues
  - updated_at OR pushed_at -> last_updated_at
If any required field is missing, include a key "error": "" in the output.
Never add extra keys.
Respond ONLY with the JSON object.
"""

# ----------------------------------------------------------------------
# Orchestration loop
# ----------------------------------------------------------------------
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
client = Anthropic(api_key="YOUR_ANTHROPIC_API_KEY")  # replace with real key

def run_adaptive_extraction_pipeline(owner: str, repo: str, max_retries: int = 3) -> GitHubRepoData:
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": f"Extract data for {owner}/{repo}."}
    ]

    for attempt in range(1, max_retries + 1):
        response = client.messages.create(
            model="claude-3-5-sonnet-20240620",
            max_tokens=500,
            temperature=0,
            messages=messages,
        )
        # Detect tool use (simplified for illustration)
        tool_use = None
        for block in response.content:
            if block.type == "tool_use":
                tool_use = block
                break

        if tool_use:
            args = tool_use.input
            if tool_use.name == "get_github_repo_data":
                api_result = get_github_repo_data(args["owner"], args["repo"])
                messages.append({"role": "assistant", "content": json.dumps(api_result)})
                continue  # give Claude the raw JSON and loop again

        # Assume final assistant message is JSON text
        agent_output = response.content[0].text.strip()
        try:
            validated = validate_agent_output(agent_output)
            logging.info("Validation succeeded on attempt %s", attempt)
            return validated
        except ValueError as ve:
            logging.warning("Validation failed: %s", ve)
            messages.append({
                "role": "user",
                "content": f"The previous JSON did not match the schema: {ve}. Please re‑format."
            })

    raise RuntimeError("Maximum retries exceeded while extracting repository data.")

# ----------------------------------------------------------------------
# Entry point
# ----------------------------------------------------------------------
if __name__ == "__main__":
    try:
        repo_data = run_adaptive_extraction_pipeline("python", "cpython")
        print(repo_data.json(indent=2))
    except Exception as e:
        logging.error("Pipeline failed: %s", e)

Expected Output

Running the script prints a nicely formatted JSON that conforms to the GitHubRepoData model, for example:

{
  "name": "cpython",
  "description": "The Python programming language",
  "stars": 77728,
  "forks": 38310,
  "open_issues": 9427,
  "last_updated_at": "2026-10-22T14:35:12Z"
}

Limitations and Tradeoffs

This approach assumes the LLM can reliably map alternate field names to the schema. If the API introduces deeply nested structures or completely new objects, the prompt would need to be expanded. The retry loop caps at three attempts; in a high‑latency environment you might increase that limit, at the cost of longer run times. Finally, the solution depends on Claude’s availability and cost; for extremely high‑throughput pipelines a pure Python parser may be cheaper.

Frequently Asked Questions

Can I use a different LLM provider?

Yes. The orchestration code is thin; replace the Anthropic client with OpenAI, Cohere, or any provider that supports tool use, adjusting the request/response handling accordingly.

What if the API returns a paginated list instead of a single object?

Wrap the pagination logic inside get_github_repo_data. Claude will still see a single consolidated dict, so the extraction prompt remains unchanged.

How do I persist the extracted data?

After validation you have a Pydantic model; you can call .dict() and write to a database, a Parquet file, or stream to a message queue. The model guarantees type safety before any persistence step.

Is the exponential backoff configuration safe for GitHub’s rate limits?

GitHub allows 60 unauthenticated requests per hour. The backoff starts at 2 seconds and caps at 10 seconds, which keeps us well under the limit for occasional spikes. For production you should use an authenticated token to raise the limit to 5 000 requests per hour.

Do I need to expose the tool to the LLM manually?

When using Claude Code Plugins, you register get_github_repo_data as a tool in the client configuration. The snippet above assumes that registration is handled by the Anthropic SDK; see the official docs for the exact registration syntax.

What I'd Change

If I were to scale this pattern across dozens of external services, I would extract the tool‑registration boilerplate into a generic APIExecutor class and drive the schema prompts from a YAML catalog. That way a new endpoint becomes a one‑line entry: define the endpoint, list possible field aliases, and the framework auto‑generates the Claude prompt and Pydantic model. The extra indirection adds a tiny startup cost but pays off in maintainability when APIs evolve en masse.

Post a Comment

Hi! How can we help you? Send us a message and we'll get back to you.