How Agentic Python Agents Can Auto‑Generate Ontologies for Nepal Finance APIs

When you need to stitch together dozens of heterogeneous Nepal‑finance JSON endpoints, writing a custom parser and a brittle if/else router for each service quickly becomes a maintenance nightmare. This post shows you how a tiny agent—built on a dynamic Pydantic model (the auto‑ontology) and a self‑registering skill registry—can discover the data contract at runtime, pick the right processing routine, and produce a verified report without any hard‑coded routing logic.

Key Takeaways

• Dynamic Pydantic models let you infer a schema from the first payload, turning an unknown JSON feed into a typed object you can validate.
• A decorator‑based skill registry decouples task logic from routing, so adding a new capability is as simple as defining a function.
• The agent can fetch, validate, route, and output results with just a few lines of reusable code, dramatically reducing time‑to‑value for new finance APIs.

Data and Sources

The demonstration uses the public JSONPlaceholder posts endpoint as a stand‑in for a Nepal‑finance transaction feed. Each record contains userId, id, title, and body, which we treat as mock transaction fields.

Source reviewed on 2026-11-13.

Agentic Python Agents with Auto‑Ontology for Nepal Finance

At the heart of the agent is an auto‑ontology step: the first record retrieved from the API is inspected, and a Pydantic BaseModel is created on the fly. This model captures field names and Python types, providing two immediate benefits:

• Validation: every subsequent record is checked against the model, catching schema drift (e.g., missing userId) before downstream code runs.
• Discoverability: the presence of particular fields (like userId) becomes a signal that a certain skill is applicable.

The agent then queries a skill registry. Each skill is a pure function that declares what it expects (a list of validated objects) and returns a serialisable result. Because the registry is a plain dictionary, adding or swapping a skill never touches the routing logic.

Implementing a Skill Framework

The framework consists of three lightweight pieces:

1. @skill(name) decorator – registers the wrapped function in a global _SKILL_REGISTRY.
2. Ontology builder – uses pydantic.create_model to generate a model from a sample payload.
3. Router – a heuristic that inspects the generated model’s fields and selects the appropriate skill.

In the example, the presence of userId triggers the aggregate_by_user skill, which counts how many posts each user posted. The pattern scales: you can add a summarize_titles skill that extracts keywords, or a detect_anomalies skill that runs statistical checks, simply by decorating a new function.

Limitations and Tradeoffs

While the auto‑ontology + skill registry approach removes hard‑coded routing, it has clear boundaries:

• Schema changes: If the API adds a new mandatory field, the dynamic model will accept it (as an extra attribute) but existing skills may ignore it, potentially hiding drift. You must monitor ValidationError logs and update skills accordingly.
• Performance: Dynamic model creation is cheap (microseconds) but validating thousands of records incurs overhead. For high‑throughput pipelines you may pre‑compile the model after the first successful fetch.
• Complex routing logic: The simple heuristic (“if field X exists, use skill Y”) works for demonstration but real finance APIs often require multi‑dimensional routing (e.g., based on transaction_type and currency). Extending the router to a rule engine is straightforward but adds complexity.
• Error handling: Network timeouts or non‑200 responses abort the run. In production you would add retries, exponential back‑off, and alerting.

Frequently Asked Questions

Do I need an LLM to drive the agent?

No. The demo relies on deterministic field matching. An LLM can replace the heuristic router for ambiguous natural‑language intents, but it introduces latency and cost.

Can this pattern handle incremental loads?

Yes. Because the ontology is derived from the first successful payload, you can store the generated Pydantic model (e.g., via pickle) and reuse it for subsequent incremental fetches, only re‑validating new fields when they appear.

What happens if the API adds a new column?

The dynamic model will accept the new key, but any skill that does not reference it will ignore it. If the new column is required for a downstream calculation, you must add a new skill or extend an existing one. Validation errors surface only when required fields are missing.

Is the skill registry thread‑safe?

In the single‑process script it is safe. For a multi‑threaded service you would protect the registry with a lock or perform registration at import time only.

How does this differ from the metaprogramming decorator post?

The earlier post focused on self‑validating handlers for a Todo API. Here we extend the idea to a full agent that not only validates but also discovers the data contract and routes tasks without any hard‑coded if/else branching, making it suitable for heterogeneous finance feeds.

Complete Script

#!/usr/bin/env python3
"""Auto‑ontology + skill registry demo for mock Nepal finance JSON feed."""

import json
import sys
from typing import Any, Dict, List, Type

import requests
from pydantic import BaseModel, ValidationError, create_model

# Global skill registry
_SKILL_REGISTRY: Dict[str, Any] = {}

def skill(name: str):
    """Decorator to register a function as a skill."""
    def decorator(fn):
        _SKILL_REGISTRY[name] = fn
        return fn
    return decorator

def fetch_posts(url: str, timeout: int = 5) -> List[Dict[str, Any]]:
    try:
        resp = requests.get(url, timeout=timeout)
        resp.raise_for_status()
    except requests.RequestException as e:
        raise RuntimeError(f"Failed to fetch data: {e}") from e
    data = resp.json()
    if not isinstance(data, list):
        raise ValueError("Expected a list of objects from the API")
    return data

def build_ontology(sample: Dict[str, Any]) -> Type[BaseModel]:
    """Create a Pydantic model from the first record."""
    fields = {k: (type(v), ...) for k, v in sample.items()}
    # pydantic expects type hints; use generic Any if type is None
    return create_model('DynamicRecord', **fields)

def validate_records(model: Type[BaseModel], records: List[Dict[str, Any]]) -> List[BaseModel]:
    """Validate each record against the dynamic model, skipping bad rows."""
    valid: List[BaseModel] = []
    for rec in records:
        try:
            valid.append(model(**rec))
        except ValidationError as e:
            print(f"Record validation failed: {e}", file=sys.stderr)
    return valid

@skill('aggregate_by_user')
def aggregate_by_user(validated: List[BaseModel]) -> List[Dict[str, Any]]:
    """Count posts per userId."""
    counter: Dict[int, int] = {}
    for item in validated:
        uid = getattr(item, 'userId')
        counter[uid] = counter.get(uid, 0) + 1
    return [{'userId': uid, 'post_count': cnt} for uid, cnt in sorted(counter.items())]

def select_skill(model: Type[BaseModel]) -> Any:
    """Simple heuristic router based on model fields."""
    if 'userId' in model.__fields__:
        return _SKILL_REGISTRY.get('aggregate_by_user')
    raise RuntimeError("No suitable skill found for the given ontology")

def print_table(rows: List[Dict[str, Any]]) -> None:
    """Render a plain‑text table to stdout."""
    if not rows:
        print("No data to display.")
        return
    headers = rows[0].keys()
    line = ' | '.join(headers)
    print(line)
    print('-' * len(line))
    for row in rows:
        print(' | '.join(str(row[h]) for h in headers))

def main() -> None:
    url = "https://jsonplaceholder.typicode.com/posts"
    raw = fetch_posts(url)
    ontology = build_ontology(raw[0])
    validated = validate_records(ontology, raw)
    skill_fn = select_skill(ontology)
    if not skill_fn:
        raise RuntimeError("Skill function not found")
    result = skill_fn(validated)
    print_table(result)
    # Save verification report
    with open("agent_report.json", "w", encoding="utf-8") as f:
        json.dump(result, f, ensure_ascii=False, indent=2)

if __name__ == "__main__":
    main()

Recommendation and Next Step

Adopt the auto‑ontology + skill registry pattern for any new Nepal‑finance JSON endpoint you need to ingest. It gives you immediate schema validation and a plug‑in point for domain‑specific logic without scattering if/else blocks throughout your codebase. As a concrete next step, clone the script above, replace the placeholder URL with a real Nepal‑finance API, and add a second skill (e.g., detect_anomalies) to start building a reusable, production‑ready data‑ingestion framework.