How to Add a New AI Agent Parser to Uber's ADR Sensor: A Step-by-Step Guide

Create a Python class inheriting from BaseParser, implement the parse_all() method to return List[AgentEvent], and expose it in adr_sensor/parsers/__init__.py for automatic discovery.

Adding an AI agent parser to Uber's ADR Sensor allows the system to ingest telemetry from custom agents that produce their own log formats. The ADR Sensor discovers parsers automatically through class inheritance, making the extension process straightforward for developers familiar with Python's abstract base classes. This guide walks through implementing, exposing, and testing a new parser based on the actual source code in uber/ADR.

Understanding the Parser Architecture

The ADR Sensor uses a plugin-style architecture where all parsers derive from a single abstract base class. In adr_sensor/parsers/base_parser.py, the BaseParser class defines the contract that every parser must fulfill:

• Implement parse_all(self) → List[AgentEvent] to extract and transform log entries
• Return standardized AgentEvent objects defined in adr_sensor/schemas/agent_event_schema.py

The Observer class in adr_sensor/observer.py handles runtime discovery—it introspects all imported subclasses of BaseParser and invokes each one during the collection phase. This means no central registry edits are required when you add a new parser.

Step 1: Create the Parser Class

Create a new file in Sensor/adr_sensor/parsers/ with a descriptive name for your agent. The file should import BaseParser, subclass it, and implement parse_all().

# Sensor/adr_sensor/parsers/myagent_parser.py

from .base_parser import BaseParser
from ..schemas.agent_event_schema import AgentEvent
from typing import List
import json

class MyAgentParser(BaseParser):
    """Parser for MyAgent-X log files."""

    def parse_all(self) -> List[AgentEvent]:
        events: List[AgentEvent] = []
        with open("/var/log/myagent_x.log") as f:
            for line in f:
                data = json.loads(line)
                events.append(
                    AgentEvent(
                        timestamp=data["ts"],
                        agent_id=data["agent_id"],
                        activity=data["activity"],
                        metadata=data.get("metadata", {}),
                    )
                )
        return events

Key implementation details:

• The AgentEvent constructor accepts timestamp, agent_id, activity, and optional metadata
• Handle missing fields gracefully using .get() with defaults
• The method must return a List[AgentEvent] even if empty—never None

Step 2: Expose the Parser for Auto-Discovery

Python's import system must see your class before Observer can discover it. Add an explicit import to the parsers package initialization file:

# Sensor/adr_sensor/parsers/__init__.py

from .myagent_parser import MyAgentParser   # ← new line

The Observer iterates over BaseParser.__subclasses__(), so the import statement is critical. Without this line, your parser exists on disk but remains invisible to the sensor at runtime.

Step 3: (Optional) Add a CLI Flag for Manual Invocation

For debugging or forced parser selection, extend adr_sensor/cli.py:

# Example addition to CLI argument parsing

parser.add_argument("--parser", choices=["myagent", "legacy", "all"], default="all")

Map the string "myagent" to MyAgentParser in your CLI handler. This step is not required for normal operation—the auto-discovery mechanism works without CLI modifications.

Step 4: Write Comprehensive Tests

Create a dedicated test file that validates your parser against synthetic data:

# Sensor/tests/test_myagent_parser.py

import json
import tempfile
from adr_sensor.parsers.myagent_parser import MyAgentParser

def test_parse_all_extracts_events():
    log_lines = [
        json.dumps({"ts": 1715424000, "agent_id": "agent-1", "activity": "inference"}),
        json.dumps({"ts": 1715424001, "agent_id": "agent-2", "activity": "training", "metadata": {"epoch": 3}}),
    ]
    
    with tempfile.NamedTemporaryFile(mode="w", suffix=".log", delete=False) as f:
        for line in log_lines:
            f.write(line + "\n")
        temp_path = f.name

    # Monkey-patch the path for testing

    parser = MyAgentParser()
    parser.__class__._test_path = temp_path  # or use dependency injection

    
    events = parser.parse_all()
    
    assert len(events) == 2
    assert events[0].agent_id == "agent-1"
    assert events[1].metadata == {"epoch": 3}

The existing tests/test_parsers.py verifies that every parser implements parse_all()—run it to confirm your addition doesn't break the generic interface contract:

cd Sensor
pytest tests/test_parsers.py -v

Step 5: Run the Full Test Suite

Before submitting changes, validate your parser in the complete test environment:

cd Sensor
pytest

Expect green results across:

• Your new unit tests
• The generic parser interface tests
• Any integration tests that exercise the Observer discovery mechanism

Using Your New Parser

Direct programmatic use:

from adr_sensor.parsers.myagent_parser import MyAgentParser

parser = MyAgentParser()
events = parser.parse_all()
for ev in events:
    print(ev.json())   # AgentEvent provides a .json() helper for serialization

Via the sensor CLI (auto-discovery enabled):

python -m adr_sensor.cli collect

The Observer will instantiate MyAgentParser alongside all other discovered parsers and aggregate their output.

Key Files Reference

Path	Purpose
adr_sensor/parsers/base_parser.py	Abstract BaseParser interface—review before implementing
adr_sensor/parsers/__init__.py	Import location for exposing new parsers
adr_sensor/parsers/myagent_parser.py	Your new parser implementation (create this)
adr_sensor/observer.py	Automatic discovery and execution of parser subclasses
adr_sensor/schemas/agent_event_schema.py	AgentEvent dataclass/structure definition
tests/test_parsers.py	Generic interface validation tests

Summary

• Inherit from BaseParser and implement parse_all() to add an AI agent parser to the ADR Sensor
• Return List[AgentEvent] with properly populated fields for downstream processing
• Import in __init__.py to enable automatic discovery by the Observer
• Test with synthetic logs covering normal cases, missing fields, and malformed entries
• No core changes required—the plugin architecture isolates your parser implementation

Frequently Asked Questions

What happens if I forget to add the import to __init__.py?

The parser class will not appear in BaseParser.__subclasses__(), so the Observer will skip it during collection. The file exists and is valid Python, but the sensor remains unaware of it. Always verify with a quick python -c "from adr_sensor.parsers import MyAgentParser; print('OK')" before running the full sensor.

Can multiple parsers read the same log file?

Yes—nothing prevents multiple BaseParser subclasses from targeting identical paths. However, implementations should avoid file-locking conflicts. For concurrent access, consider copying or memory-mapping logs, or implement file locking in your parse_all() method.

How do I update the parser for a changed log format?

Modify your parse_all() implementation to handle both old and new formats, or version your parser by creating MyAgentV2Parser. The sensor will execute both if both are importable. Deprecate old parsers by removing their __init__.py import after confirming no production systems require the legacy format.

Does the ADR Sensor support real-time log streaming?

The base BaseParser interface uses parse_all(), which implies batch processing of complete files. For true streaming, implement a generator-based approach within parse_all() that yields events as they arrive, or propose an upstream contribution to uber/ADR that extends the interface with parse_stream().

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}