Class ActionEvent

app/api/agentops/exporter/models.py:292–359

Action becomes a (child) Span

class ActionEvent(BaseModel):
    """
    Action becomes a (child) Span
    """

    id: UUID
    session_id: Optional[UUID] = None
    agent_id: Optional[UUID] = None
    action_type: Optional[str] = None
    logs: Optional[str] = None
    screenshot: Optional[str] = None
    params: Optional[str] = None
    returns: Optional[str] = None
    init_timestamp: Optional[datetime] = None
    end_timestamp: Optional[datetime] = None

    async def to_span(
        self,
        trace_id: Optional[str] = None,
        parent_span_id: Optional[str] = None,
        project_id: Optional[str] = None,
    ) -> Span:
        """Convert an ActionEvent to a Span"""
        span_attributes = {
            "action.id": str(self.id),
            "action.type": self.action_type or "",
            "session.id": str(self.session_id),
            "agent.id": str(self.agent_id),
        }

        if self.params:
            if isinstance(self.params, dict):
                span_attributes["action.params"] = json.dumps(self.params)
            else:
                span_attributes["action.params"] = str(self.params)

        if self.returns:
            if isinstance(self.returns, dict):
                span_attributes["action.returns"] = json.dumps(self.returns)
            else:
                span_attributes["action.returns"] = str(self.returns)

        if self.screenshot:
            span_attributes["action.screenshot"] = self.screenshot

        init_timestamp_ns = datetime_to_ns(self.init_timestamp)
        events = []
        if self.logs:
            events.append(
                {"timestamp": init_timestamp_ns, "name": "logs", "attributes": {"log.message": self.logs}}
            )

        return Span(
            name=f"action:{self.action_type or 'unknown'}",
            trace_id=trace_id,
            span_id=str(self.id),
            parent_span_id=parent_span_id,
            kind=SpanKind.INTERNAL,