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

> Learn how to add a new AI agent parser to Uber's ADR Sensor. Follow this guide to implement parse_all and register your parser for automatic discovery.

- Repository: [Uber Open Source/ADR](https://github.com/uber/ADR)
- Tags: how-to-guide
- Published: 2026-08-07

---

**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`](https://github.com/uber/ADR/blob/main/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`](https://github.com/uber/ADR/blob/main/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`](https://github.com/uber/ADR/blob/main/adr_sensor/schemas/agent_event_schema.py)

The **`Observer`** class in [`adr_sensor/observer.py`](https://github.com/uber/ADR/blob/main/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()`.

```python

# 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:

```python

# 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`](https://github.com/uber/ADR/blob/main/adr_sensor/cli.py):

```python

# 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:

```python

# 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`](https://github.com/uber/ADR/blob/main/tests/test_parsers.py) verifies that every parser implements `parse_all()`—run it to confirm your addition doesn't break the generic interface contract:

```bash
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:

```bash
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:**

```python
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):**

```bash
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`](https://github.com/uber/ADR/blob/main/adr_sensor/parsers/base_parser.py) | Abstract `BaseParser` interface—review before implementing |
| [`adr_sensor/parsers/__init__.py`](https://github.com/uber/ADR/blob/main/adr_sensor/parsers/__init__.py) | Import location for exposing new parsers |
| [`adr_sensor/parsers/myagent_parser.py`](https://github.com/uber/ADR/blob/main/adr_sensor/parsers/myagent_parser.py) | Your new parser implementation (create this) |
| [`adr_sensor/observer.py`](https://github.com/uber/ADR/blob/main/adr_sensor/observer.py) | Automatic discovery and execution of parser subclasses |
| [`adr_sensor/schemas/agent_event_schema.py`](https://github.com/uber/ADR/blob/main/adr_sensor/schemas/agent_event_schema.py) | `AgentEvent` dataclass/structure definition |
| [`tests/test_parsers.py`](https://github.com/uber/ADR/blob/main/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`](https://github.com/uber/ADR/blob/main/__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`](https://github.com/uber/ADR/blob/main/__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`](https://github.com/uber/ADR/blob/main/__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()`.