feat(redaction): add event-level redact_event hook

README.md

@@ -104,6 +104,27 @@ def redact_sync(text):
 agentcat.track(server, "proj_0000000", AgentCatOptions(redact_sensitive_information=redact_sync))
 ```
 
+For redaction decisions that need more context than a single string — such as which tool was called or what type of event is being published — use the event-level `redact_event` hook. It receives the full event object and returns a modified event, or `None` to drop the event entirely. It may be sync or async, and can be combined with `redact_sensitive_information`.
+
+```python
+from agentcat import AgentCatOptions
+
+def redact_event(event):
+    # Drop events from tools that handle secrets entirely
+    if event.resource_name == "get_credentials":
+        return None
+    # Strip response payloads from a specific tool
+    if event.resource_name == "export_report":
+        modified = event.model_copy()
+        modified.response = None
+        return modified
+    return event
+
+agentcat.track(server, "proj_0000000", AgentCatOptions(redact_event=redact_event))
+```
+
+When both hooks are configured, `redact_event` runs first and sees the raw, unredacted values; `redact_sensitive_information` then runs on its output as a final string-level scrub. The system-managed fields `id`, `session_id`, `project_id`, `event_type`, and `timestamp` cannot be changed by the hook, and if the hook raises, the event is dropped.
+
 ### Forwarding data to existing observability platforms
 
 AgentCat seamlessly integrates with your existing observability stack, providing automatic logging and tracing without the tedious setup typically required. Export telemetry data to multiple platforms simultaneously:

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "agentcat"
-version = "1.0.2"
+version = "1.0.3"
 description = "Analytics tool for MCP (Model Context Protocol) servers and AI agents - tracks tool usage patterns and provides insights"
 authors = [
     { name = "AgentCat, Inc.", email = "support@agentcat.com" },

src/agentcat/__init__.py

@@ -22,7 +22,9 @@
 from .modules.internal import set_server_tracking_data
 from .modules.logging import set_debug_mode, write_to_log
 from .types import (
+    Event,
     EventPropertiesFunction,
+    EventRedactionFunction,
     EventTagsFunction,
     IdentifyFunction,
     AgentCatData,
@@ -256,8 +258,10 @@ def _apply_server_tracking(
     # Types for identify functionality
     "UserIdentity",
     "IdentifyFunction",
-    # Type for redaction functionality
+    # Types for redaction functionality
     "RedactionFunction",
+    "EventRedactionFunction",
+    "Event",
     # Types for event metadata callbacks
     "EventTagsFunction",
     "EventPropertiesFunction",

src/agentcat/modules/event_queue.py

@@ -21,7 +21,7 @@
 from .compatibility import get_mcp_compatible_error_message
 from .internal import get_server_tracking_data
 from .logging import write_to_log
-from .redaction import redact_event
+from .redaction import apply_event_redaction, redact_event
 from .sanitization import sanitize_event
 from .truncation import truncate_event
 from .session import get_session_info, set_last_activity
@@ -105,6 +105,23 @@ def _worker(self) -> None:
 
     def _process_event(self, event: UnredactedEvent) -> None:
         """Process a single event."""
+        if event and event.redact_event_fn:
+            # Run the customer's event-level redaction hook first, on raw values
+            event_redact_fn = event.redact_event_fn
+            event.redact_event_fn = None
+            try:
+                if not apply_event_redaction(event, event_redact_fn):
+                    write_to_log(
+                        f"Event {event.id or 'unknown'} dropped by redact_event hook"
+                    )
+                    return
+            except Exception as error:
+                write_to_log(
+                    f"Failed to redact event (event-level hook), dropping event "
+                    f"{event.id or 'unknown'}: {error}"
+                )
+                return
+
         if event and event.redaction_fn:
             # Redact sensitive information if a redaction function is provided
             try:
@@ -320,6 +337,7 @@ def publish_event(server: Any, event: UnredactedEvent) -> None:
     full_event = UnredactedEvent(
         **merged_data,
         redaction_fn=data.options.redact_sensitive_information,
+        redact_event_fn=data.options.redact_event,
     )
 
     set_last_activity(server)

src/agentcat/modules/redaction.py

@@ -1,9 +1,23 @@
 """PII redaction for AgentCat logs."""
 
+import asyncio
+import inspect
 from typing import Any, TYPE_CHECKING, Callable, Set
 
 if TYPE_CHECKING:
-    from agentcat.types import Event, UnredactedEvent
+    from agentcat.types import Event, EventRedactionFunction, UnredactedEvent
+
+
+# System-managed fields restored after the event-level redaction hook runs.
+# These are required for ingestion and session/project attribution, so
+# consumer changes to them are ignored.
+RESTORED_FIELDS = (
+    "id",
+    "session_id",
+    "project_id",
+    "event_type",
+    "timestamp",
+)
 
 
 # Set of field names that should be protected from redaction.
@@ -101,3 +115,48 @@ def redact_event(event: "UnredactedEvent", redact_fn: Callable[[str], str]) -> "
         A new event object with all strings redacted
     """
     return redact_strings_in_object(event, redact_fn, "", False)
+
+
+def apply_event_redaction(
+    event: "UnredactedEvent", event_redact_fn: "EventRedactionFunction"
+) -> bool:
+    """
+    Applies the customer's event-level redaction hook to an event, in place.
+    The hook receives the full event (without internal function fields) and may
+    return a modified event, or None to drop the event entirely.
+
+    The event object is rewritten rather than replaced: the queue pipeline
+    processes the same object across steps, and copying every field ensures
+    values the hook cleared stay cleared. System-managed fields are restored
+    from the original, and the string-level redaction_fn is preserved so it
+    still runs afterwards.
+
+    Args:
+        event: The event to run the hook on; mutated with the hook's result
+        event_redact_fn: The customer's event-level redaction hook
+
+    Returns:
+        True if the event was kept, False if the hook dropped it
+    """
+    from agentcat.types import Event
+
+    # Hide internal carrier fields from the hook
+    hook_input = Event(**{name: getattr(event, name) for name in Event.model_fields})
+
+    result = event_redact_fn(hook_input)
+    if inspect.iscoroutine(result):
+        # The queue worker runs in a plain thread with no event loop
+        result = asyncio.run(result)
+
+    if result is None:
+        return False
+
+    # Copy every field from the result so values the hook cleared stay
+    # cleared, then restore the system-managed fields from the original
+    snapshot = {name: getattr(event, name) for name in RESTORED_FIELDS}
+    for name in Event.model_fields:
+        setattr(event, name, getattr(result, name, None))
+    for name, value in snapshot.items():
+        setattr(event, name, value)
+
+    return True

src/agentcat/types.py

@@ -14,6 +14,13 @@
 IdentifyFunction = Callable[[dict[str, Any], Any], Optional["UserIdentity"]]
 # Type alias for redaction function
 RedactionFunction = Callable[[str], str | Awaitable[str]]
+# Type alias for the event-level redaction hook — receives the full Event and
+# returns a modified Event, or None to drop the event entirely.
+# Accepts sync or async callables (mirrors RedactionFunction).
+EventRedactionFunction = Callable[
+    ["Event"],
+    Union["Event", None, Awaitable[Union["Event", None]]],
+]
 # Type alias for event_tags callback — returns str:str map attached to every auto-captured event.
 # Accepts sync or async callables (mirrors RedactionFunction).
 EventTagsFunction = Callable[
@@ -117,6 +124,7 @@ class EventType(str, Enum):
 
 class UnredactedEvent(Event):
     redaction_fn: RedactionFunction | None = None
+    redact_event_fn: EventRedactionFunction | None = None  # Whole-event redaction hook
 
 
 @dataclass
@@ -176,6 +184,16 @@ class AgentCatOptions:
     custom_context_description: str = DEFAULT_CONTEXT_DESCRIPTION
     identify: IdentifyFunction | None = None
     redact_sensitive_information: RedactionFunction | None = None
+    # Event-level redaction hook invoked with the full event (inspect
+    # resource_name, event_type, parameters, response, etc.) before it is
+    # published. Return a modified event, or None to drop the event entirely.
+    # May be sync or async. Runs before redact_sensitive_information, so it
+    # sees raw, unredacted values; the string-level hook, sanitization, and
+    # truncation still run on its output. The system-managed fields id,
+    # session_id, project_id, event_type, and timestamp cannot be changed
+    # (id may be None at hook time). If the hook raises, the event is dropped
+    # and the error is logged to ~/agentcat.log.
+    redact_event: EventRedactionFunction | None = None
     exporters: dict[str, ExporterConfig] | None = None
     debug_mode: bool = False
     api_base_url: str | None = None

tests/e2e/official/test_redact_event_http.py

@@ -0,0 +1,124 @@
+"""Event-level redaction hook (redact_event) over real-wire payloads.
+
+Unlike the string-level hook (see test_redaction_http.py, xfail-tracked), the
+event-level hook receives the Pydantic event object directly, so it works on
+the live publish path today.
+"""
+
+from __future__ import annotations
+
+import time
+
+import pytest
+from mcp import ClientSession
+from mcp.client.streamable_http import streamablehttp_client
+
+from agentcat.modules.internal import get_server_tracking_data
+
+
+pytestmark = pytest.mark.e2e
+
+
+def _set_redact_event(server, fn) -> None:
+    data = get_server_tracking_data(server)
+    assert data is not None
+    data.options.redact_event = fn
+
+
+@pytest.mark.asyncio
+async def test_hook_inspects_metadata_and_rewrites_event_fields(
+    official_http_server, capture_queue
+):
+    url, server = official_http_server
+
+    seen_resource_names: list[str | None] = []
+
+    def hook(event):
+        seen_resource_names.append(event.resource_name)
+        if event.resource_name == "add_todo":
+            modified = event.model_copy()
+            modified.parameters = {"replaced": True}
+            return modified
+        return event
+
+    _set_redact_event(server, hook)
+    try:
+        async with streamablehttp_client(url) as (read, write, _):
+            async with ClientSession(read, write) as client:
+                await client.initialize()
+                await client.call_tool(
+                    "add_todo", {"text": "buy milk", "context": "rewrite"}
+                )
+
+        time.sleep(0.5)
+        call_events = [e for e in capture_queue if e.event_type == "mcp:tools/call"]
+        assert call_events
+        assert "add_todo" in seen_resource_names
+        assert call_events[0].parameters == {"replaced": True}
+
+        # System-managed fields are intact
+        assert call_events[0].session_id
+        assert call_events[0].event_type == "mcp:tools/call"
+    finally:
+        _set_redact_event(server, None)
+
+
+@pytest.mark.asyncio
+async def test_hook_returning_none_drops_the_event(
+    official_http_server, capture_queue
+):
+    url, server = official_http_server
+
+    def hook(event):
+        if event.event_type == "mcp:tools/call":
+            return None
+        return event
+
+    _set_redact_event(server, hook)
+    try:
+        async with streamablehttp_client(url) as (read, write, _):
+            async with ClientSession(read, write) as client:
+                await client.initialize()
+                await client.call_tool("add_todo", {"text": "drop", "context": "drop"})
+                await client.list_tools()
+
+        time.sleep(1.0)
+        call_events = [e for e in capture_queue if e.event_type == "mcp:tools/call"]
+        assert not call_events, (
+            f"redact_event returning None must drop the event, got {len(call_events)}"
+        )
+        # Other event types still publish
+        assert capture_queue
+    finally:
+        _set_redact_event(server, None)
+
+
+@pytest.mark.asyncio
+async def test_hook_raising_drops_only_the_affected_event(
+    official_http_server, capture_queue
+):
+    url, server = official_http_server
+
+    def hook(event):
+        if event.event_type == "mcp:tools/call":
+            raise RuntimeError("hook exploded")
+        return event
+
+    _set_redact_event(server, hook)
+    try:
+        async with streamablehttp_client(url) as (read, write, _):
+            async with ClientSession(read, write) as client:
+                await client.initialize()
+                await client.call_tool("add_todo", {"text": "boom", "context": "boom"})
+                # Server still serves requests after the hook error
+                tools = await client.list_tools()
+                assert tools.tools
+
+        time.sleep(1.0)
+        call_events = [e for e in capture_queue if e.event_type == "mcp:tools/call"]
+        assert not call_events, (
+            f"redact_event raising must drop the event, got {len(call_events)}"
+        )
+        assert capture_queue  # other events still arrived
+    finally:
+        _set_redact_event(server, None)