# How to Filter ADR Agent Events by Source (Claude, Cursor, Codex)

> Learn to filter ADR Agent Events by source like Claude, Cursor, or Codex. Use Python list comprehensions to easily manage AgentEvent objects based on their source attribute.

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

---

**Filter Uber ADR `AgentEvent` objects by checking the `source` attribute against agent identifiers like `"claude"`, `"cursor"`, or `"codex"` using standard Python list comprehensions.**

The Uber ADR Sensor library captures telemetry from AI coding agents and represents each record as an `AgentEvent` dataclass. Since the `source` field stores the originating agent identifier, you can filter ADR agent events by source with simple Python collection operations—no complex query language required.

---

## Understanding the `source` Field in AgentEvent

The `AgentEvent` dataclass in [`Sensor/adr_sensor/schemas/agent_event_schema.py`](https://github.com/uber/ADR/blob/main/Sensor/adr_sensor/schemas/agent_event_schema.py) defines a `source` attribute at line 49 that stores the agent identifier as a plain string:

```python

# From adr_sensor/schemas/agent_event_schema.py (line 49)

@dataclass(frozen=True)
class AgentEvent:
    source: str           # "claude", "cursor", "codex", etc.

    # ... other fields

```

Because `@dataclass(frozen=True)` makes the schema immutable, the `source` value is set once during event creation and cannot be modified afterward. This guarantees consistent, deterministic filtering.

Individual parsers populate this field:

- **Claude Parser** ([`claude_parser.py`](https://github.com/uber/ADR/blob/main/claude_parser.py)) sets `source="claude"`
- **Cursor Parser** ([`cursor_parser.py`](https://github.com/uber/ADR/blob/main/cursor_parser.py)) sets `source="cursor"`
- **Codex Parser** ([`codex_parser.py`](https://github.com/uber/ADR/blob/main/codex_parser.py)) sets `source="codex"`

---

## Filtering Events After Ingestion

The `AgentObserver` API provides `ingest_all()` to load events from log directories. Apply a **list comprehension** to filter by source:

```python
from adr_sensor import AgentObserver

# Load all events from the default log directory

observer = AgentObserver()
all_events, _ = observer.ingest_all()  # Returns (list[AgentEvent], list[SystemConfig])

# Filter for Claude, Cursor, or Codex events

allowed_sources = {"claude", "cursor", "codex"}
filtered_events = [e for e in all_events if e.source in allowed_sources]

print(f"Total events: {len(all_events)}")
print(f"Filtered events: {len(filtered_events)}")

# Display summary of filtered results

observer.display_summary(filtered_events, [])

```

This approach works because `AgentEvent` objects are plain Python dataclasses with public attributes.

---

## Reusable Filtering Function

For scripts and notebooks, extract the logic into a reusable function:

```python
def filter_events_by_source(events, sources):
    """
    Return AgentEvent objects whose source matches any entry in sources.
    
    Parameters:
        events: list[AgentEvent] - events to filter
        sources: list[str] or set[str] - agent identifiers to keep
    
    Returns:
        list[AgentEvent] - filtered events
    """
    source_set = set(sources)  # O(1) lookup

    return [e for e in events if e.source in source_set]


# Usage example

from adr_sensor import AgentObserver

observer = AgentObserver()
events, _ = observer.ingest_all()

# Filter for specific agents

target_agents = ["claude", "cursor", "codex"]
filtered = filter_events_by_source(events, target_agents)

# Pass to downstream analysis or export

observer.display_summary(filtered, [])

```

Using `set(sources)` ensures **O(1)** membership testing for large event volumes.

---

## Filtering Custom Event Collections

If you generate events outside the standard ingestion pipeline—for example, using the demo utilities in [`Sensor/examples/demo.py`](https://github.com/uber/ADR/blob/main/Sensor/examples/demo.py)—the same pattern applies:

```python
from examples.demo import create_sample_events

# Create synthetic events with various sources

events = create_sample_events()

# Select only Claude, Cursor, and Codex

agent_events = [e for e in events if e.source in {"claude", "cursor", "codex"}]

```

This flexibility lets you integrate source filtering into custom parsers, testing suites, or data pipelines.

---

## Key Source Files

| File | Purpose |
|------|---------|
| [`Sensor/adr_sensor/schemas/agent_event_schema.py`](https://github.com/uber/ADR/blob/main/Sensor/adr_sensor/schemas/agent_event_schema.py) | Defines `AgentEvent` dataclass with `source` field (line 49) |
| [`Sensor/adr_sensor/parsers/claude_parser.py`](https://github.com/uber/ADR/blob/main/Sensor/adr_sensor/parsers/claude_parser.py) | Sets `source="claude"` during event construction |
| [`Sensor/adr_sensor/parsers/cursor_parser.py`](https://github.com/uber/ADR/blob/main/Sensor/adr_sensor/parsers/cursor_parser.py) | Sets `source="cursor"` during event construction |
| [`Sensor/adr_sensor/parsers/codex_parser.py`](https://github.com/uber/ADR/blob/main/Sensor/adr_sensor/parsers/codex_parser.py) | Sets `source="codex"` during event construction |
| [`Sensor/examples/demo.py`](https://github.com/uber/ADR/blob/main/Sensor/examples/demo.py) | Provides `create_sample_events()` for testing filters |

---

## Performance Considerations

- **Time complexity:** O(n) where n = number of events; set lookup for `source` matching is O(1)
- **Memory:** List comprehensions create new lists; for large datasets, consider generator expressions: `(e for e in events if e.source in allowed)`
- **Frozen dataclasses:** Immutable `AgentEvent` objects enable safe sharing across threads without copying

---

## Summary

- The `source` attribute in `AgentEvent` stores agent identifiers as strings: `"claude"`, `"cursor"`, `"codex"`
- Filter with standard Python: `[e for e in events if e.source in {"claude", "cursor", "codex"}]`
- `AgentObserver.ingest_all()` returns events ready for source-based filtering
- Extract reusable functions for scripts, notebooks, and automated pipelines
- Immutable schema guarantees consistent, thread-safe filtering

---

## Frequently Asked Questions

### How does the ADR Sensor know which agent created an event?

Each parser hardcodes its agent identifier. The Claude Parser sets `source="claude"`, the Cursor Parser sets `source="cursor"`, and the Codex Parser sets `source="codex"` when constructing `AgentEvent` objects. This design ensures consistent identifiers across the codebase regardless of log format variations.

### Can I filter events before ingestion instead of after?

The `AgentObserver.ingest_all()` method loads all events first. For pre-ingestion filtering, you would need to implement custom logic at the parser or file-scanning level—ADR does not currently expose built-in source filters during ingestion. Post-ingestion filtering with list comprehensions is the recommended approach.

### What happens if I pass an unknown source name to my filter?

The filter simply excludes events with non-matching sources. No error is raised. If you need to validate source names against known agents, inspect the parser files or maintain your own allowlist: `{"claude", "cursor", "codex", "copilot", "gpt-engineer", ...}`.

### Are AgentEvent objects modifiable after creation?

No. The `@dataclass(frozen=True)` decorator prevents field modification after instantiation. This immutability makes source filtering safe for concurrent processing and ensures audit trail integrity in security-sensitive deployments.