Custom events, PostHog exporter, async identify, and hardened event pipeline

README.md

@@ -72,7 +72,7 @@ agentcat.track(server, "proj_0000000")
 
 ### Identifying users
 
-You can identify your user sessions with a simple callback AgentCat exposes, called `identify`.
+You can identify your user sessions with a simple callback AgentCat exposes, called `identify`. The hook runs on every captured request and an `agentcat:identify` event is published each time it returns an identity; `user_id`/`user_name` are overwritten and `user_data` keys merge across calls.
 
 ```python
 from agentcat import AgentCatOptions, UserIdentity

src/agentcat/__init__.py

@@ -9,7 +9,11 @@
 __version__ = version("agentcat")
 
 from agentcat.modules.overrides.mcp_server import override_lowlevel_mcp_server
-from agentcat.modules.session import get_session_info, new_session_id
+from agentcat.modules.session import (
+    derive_session_id_from_mcp_session,
+    get_session_info,
+    new_session_id,
+)
 
 from .modules.compatibility import (
     COMPATIBILITY_ERROR_MESSAGE,
@@ -18,16 +22,20 @@
     is_compatible_server,
     is_official_fastmcp_server,
 )
+from .modules.constants import AGENTCAT_CUSTOM_EVENT_TYPE
 from .modules.diagnostics import init_diagnostics
-from .modules.internal import set_server_tracking_data
+from .modules.internal import get_server_tracking_data, set_server_tracking_data
 from .modules.logging import set_debug_mode, write_to_log
+from .modules.validation import validate_tags
 from .types import (
+    CustomEventData,
     EventPropertiesFunction,
     EventTagsFunction,
     IdentifyFunction,
     AgentCatData,
     AgentCatOptions,
     RedactionFunction,
+    UnredactedEvent,
     UserIdentity,
 )
 
@@ -248,9 +256,114 @@ def _apply_server_tracking(
         override_lowlevel_mcp_server(lowlevel_server, data)
 
 
+def publish_custom_event(
+    server_or_session_id: Any,
+    project_id: str,
+    event_data: CustomEventData | None = None,
+) -> None:
+    """
+    Publish a custom event to AgentCat with flexible session management.
+
+    Args:
+        server_or_session_id: Either a tracked MCP server instance or an MCP
+            session ID string. For a session ID string, a deterministic
+            AgentCat session ID is derived so the same inputs always correlate
+            to the same session.
+        project_id: Your AgentCat project ID (required)
+        event_data: Optional event data to include with the custom event
+
+    Raises:
+        ValueError: If project_id is missing, or the server is not tracked
+        TypeError: If the first parameter is neither a server nor a string
+
+    Example:
+        >>> import agentcat
+        >>> data = agentcat.CustomEventData(
+        ...     resource_name="custom-action",
+        ...     parameters={"action": "user-feedback", "rating": 5},
+        ...     message="User provided feedback",
+        ... )
+        >>> agentcat.publish_custom_event(server, "proj_abc123", data)
+    """
+    from agentcat.modules import event_queue as event_queue_module
+
+    # Normalize once: an omitted payload behaves like an all-defaults payload.
+    event_data = event_data or CustomEventData()
+
+    if not project_id:
+        raise ValueError("project_id is required for publish_custom_event")
+
+    tracked_server: Any | None = None
+
+    if isinstance(server_or_session_id, str):
+        # Custom session ID provided - derive a deterministic session ID
+        session_id = derive_session_id_from_mcp_session(
+            server_or_session_id, project_id
+        )
+    elif server_or_session_id is not None and not isinstance(
+        server_or_session_id, (int, float, bool)
+    ):
+        try:
+            data = get_server_tracking_data(server_or_session_id)
+        except TypeError:
+            # Non-weakref-able / unhashable objects can't be tracked servers;
+            # surface the documented "not tracked" error instead of leaking a
+            # WeakKeyDictionary internal TypeError.
+            data = None
+        if data is None:
+            # Server is not tracked - treat it as an error
+            raise ValueError(
+                "Server is not tracked. Please call agentcat.track() first or "
+                "provide a session ID string."
+            )
+        # Use the tracked server's session ID and configuration
+        tracked_server = server_or_session_id
+        session_id = data.session_id
+    else:
+        raise TypeError(
+            "First parameter must be either an MCP server object or a session ID string"
+        )
+
+    event = UnredactedEvent(
+        # Core fields
+        session_id=session_id,
+        project_id=project_id,
+        # Fixed event type for custom events
+        event_type=AGENTCAT_CUSTOM_EVENT_TYPE,
+        timestamp=datetime.now(timezone.utc),
+        # Event data from parameters
+        resource_name=event_data.resource_name,
+        parameters=event_data.parameters,
+        response=event_data.response,
+        user_intent=event_data.message,
+        duration=event_data.duration,
+        is_error=event_data.is_error,
+        error=event_data.error,
+    )
+
+    # Wire up customer-defined metadata
+    if event_data.tags:
+        event.tags = validate_tags(event_data.tags)
+    if event_data.properties:
+        event.properties = event_data.properties
+
+    # If we have a tracked server, publish through it (merges session info and
+    # redaction config); otherwise add directly to the event queue.
+    if tracked_server is not None:
+        event_queue_module.publish_event(tracked_server, event)
+    else:
+        event_queue_module.event_queue.add(event)
+
+    write_to_log(
+        f"Published custom event for session {session_id} with type "
+        f"'{AGENTCAT_CUSTOM_EVENT_TYPE}'"
+    )
+
+
 __all__ = [
     # Main API
     "track",
+    "publish_custom_event",
     # Configuration
     "AgentCatOptions",
     # Types for identify functionality
@@ -261,4 +374,6 @@ def _apply_server_tracking(
     # Types for event metadata callbacks
     "EventTagsFunction",
     "EventPropertiesFunction",
+    # Type for custom events
+    "CustomEventData",
 ]

src/agentcat/modules/constants.py

@@ -4,6 +4,11 @@
 EVENT_ID_PREFIX = "evt"
 AGENTCAT_API_URL = "https://api.agentcat.com"  # Default API URL for AgentCat events
 AGENTCAT_SOURCE = "agentcat"  # Source attribution for telemetry exporters
+AGENTCAT_CUSTOM_EVENT_TYPE = "agentcat:custom"  # Event type for publish_custom_event
+# Event types defined by this SDK (all "agentcat:"-prefixed) that the generated
+# API client's enum doesn't know about; types.Event lets these bypass the
+# generated enum check. Add new SDK-defined event types here.
+SDK_EVENT_TYPES = frozenset({AGENTCAT_CUSTOM_EVENT_TYPE})
 DEFAULT_CONTEXT_DESCRIPTION = "Explain why you are calling this tool and how it fits into the user's overall goal. This parameter is used for analytics and user intent tracking. YOU MUST provide 15-25 words (count carefully). NEVER use first person ('I', 'we', 'you') - maintain third-person perspective. NEVER include sensitive information such as credentials, passwords, or personal data. Example (20 words): \"Searching across the organization's repositories to find all open issues related to performance complaints and latency issues for team prioritization.\""
 
 # Maximum number of exceptions to capture in a cause chain

src/agentcat/modules/event_queue.py

@@ -20,7 +20,7 @@
 from ..utils import generate_prefixed_ksuid
 from .compatibility import get_mcp_compatible_error_message
 from .internal import get_server_tracking_data
-from .logging import write_to_log
+from .logging import safe_error_string, write_to_log
 from .redaction import redact_event
 from .sanitization import sanitize_event
 from .truncation import truncate_event
@@ -115,8 +115,10 @@ def _process_event(self, event: UnredactedEvent) -> None:
                 event = redacted_event
                 event.redaction_fn = None  # Clear the function to avoid reprocessing
             except Exception as error:
+                # safe_error_string: the error comes from the customer's
+                # redaction function and may have a broken __str__.
                 write_to_log(
-                    f"WARNING: Dropping event {event.id or 'unknown'} due to redaction failure: {error}"
+                    f"WARNING: Dropping event {event.id or 'unknown'} due to redaction failure: {safe_error_string(error)}"
                 )
                 return  # Skip this event if redaction fails
 
@@ -287,40 +289,54 @@ def set_event_queue(new_queue: EventQueue) -> None:
 
 
 def publish_event(server: Any, event: UnredactedEvent) -> None:
-    """Publish an event to the queue."""
-    if not event.duration:
-        if event.timestamp:
-            event.duration = int(
-                (datetime.now(timezone.utc).timestamp() - event.timestamp.timestamp())
-                * 1000
-            )
-        else:
-            event.duration = None
+    """Publish an event to the queue.
 
-    data = get_server_tracking_data(server)
-    if not data:
-        write_to_log(
-            "Warning: Server tracking data not found. Event will not be published."
-        )
-        return
+    Never raises: this is called from the customer's request path (tool-call
+    and initialize wrappers). Any failure degrades to a dropped event plus a
+    log line.
+    """
+    try:
+        if not event.duration:
+            if event.timestamp:
+                event.duration = int(
+                    (
+                        datetime.now(timezone.utc).timestamp()
+                        - event.timestamp.timestamp()
+                    )
+                    * 1000
+                )
+            else:
+                event.duration = None
 
-    session_info = get_session_info(server, data)
+        data = get_server_tracking_data(server)
+        if not data:
+            write_to_log(
+                "Warning: Server tracking data not found. Event will not be published."
+            )
+            return
 
-    # Create full event with all required fields
-    # Merge event data with session info
-    event_data = event.model_dump(exclude_none=True)
-    session_data = session_info.model_dump(exclude_none=True)
+        session_info = get_session_info(server, data)
 
-    # Merge data, ensuring project_id from data takes precedence
-    merged_data = {**event_data, **session_data}
-    merged_data["project_id"] = (
-        data.project_id
-    )  # Override with tracking data's project_id
+        # Create full event with all required fields
+        # Merge event data with session info
+        event_data = event.model_dump(exclude_none=True)
+        session_data = session_info.model_dump(exclude_none=True)
 
-    full_event = UnredactedEvent(
-        **merged_data,
-        redaction_fn=data.options.redact_sensitive_information,
-    )
+        # Merge data, ensuring project_id from data takes precedence
+        merged_data = {**event_data, **session_data}
+        merged_data["project_id"] = (
+            data.project_id
+        )  # Override with tracking data's project_id
 
-    set_last_activity(server)
-    event_queue.add(full_event)
+        full_event = UnredactedEvent(
+            **merged_data,
+            redaction_fn=data.options.redact_sensitive_information,
+        )
+
+        set_last_activity(server)
+        event_queue.add(full_event)
+    except Exception as e:
+        write_to_log(
+            f"WARNING: Dropping event {getattr(event, 'id', None) or 'unknown'} — "
+            f"publish failed: {safe_error_string(e)}"
+        )

src/agentcat/modules/exceptions.py

@@ -12,6 +12,9 @@
 from agentcat.types import ChainedErrorData, ErrorData, StackFrame
 from agentcat.modules.constants import MAX_EXCEPTION_CHAIN_DEPTH, MAX_STACK_FRAMES
 
+# str(value) that never raises (exceptions may have a broken __str__).
+from agentcat.modules.logging import safe_error_string as _safe_str
+
 _captured_error: contextvars.ContextVar[BaseException | None] = contextvars.ContextVar(
     "_captured_error", default=None
 )
@@ -26,12 +29,33 @@ def capture_exception(exc: BaseException | Any) -> ErrorData:
     tracebacks into structured frames and detects whether each frame is user
     code (in_app: True) or library code (in_app: False).
 
+    This function must NEVER raise — it runs on the customer's request path
+    (tool-call wrappers) where an escaping exception would corrupt the
+    customer's request handling. Any internal failure degrades to a minimal
+    ErrorData dict.
+
     Args:
         exc: The error to capture (can be BaseException, string, object, or any value)
 
     Returns:
         ErrorData dict with structured error information including platform="python"
     """
+    try:
+        return _capture_exception_unsafe(exc)
+    except Exception:
+        try:
+            type_name = type(exc).__name__ if isinstance(exc, BaseException) else None
+        except Exception:
+            type_name = None
+        return {
+            "message": _safe_str(exc),
+            "type": type_name,
+            "platform": "python",
+        }
+
+
+def _capture_exception_unsafe(exc: BaseException | Any) -> ErrorData:
+    """Implementation of capture_exception; see its docstring."""
     if is_call_tool_result(exc):
         return capture_call_tool_result_error(exc)
 
@@ -43,7 +67,7 @@ def capture_exception(exc: BaseException | Any) -> ErrorData:
         }
 
     error_data: ErrorData = {
-        "message": str(exc),
+        "message": _safe_str(exc),
         "type": type(exc).__name__,
         "platform": "python",
     }
@@ -281,7 +305,7 @@ def unwrap_exception_chain(exc: BaseException) -> list[ChainedErrorData]:
             break
 
         chained_data: ChainedErrorData = {
-            "message": str(next_exc),
+            "message": _safe_str(next_exc),
             "type": type(next_exc).__name__,
         }
 
@@ -315,7 +339,7 @@ def format_exception_string(exc: BaseException) -> str:
     try:
         return "".join(traceback.format_exception(type(exc), exc, exc.__traceback__))
     except Exception:
-        return f"{type(exc).__name__}: {exc}"
+        return f"{type(exc).__name__}: {_safe_str(exc)}"
 
 
 def is_call_tool_result(value: Any) -> bool:
@@ -403,7 +427,7 @@ def stringify_non_exception(value: Any) -> str:
 
         return json.dumps(value)
     except Exception:
-        return str(value)
+        return _safe_str(value)
 
 
 def store_captured_error(exc: BaseException) -> None: