Class ToolEvent

app/api/agentops/exporter/models.py:505–569

ToolEvent becomes a (child) Span

class ToolEvent(BaseModel):
    """
    ToolEvent becomes a (child) Span
    """

    id: UUID
    session_id: Optional[UUID] = None
    agent_id: Optional[UUID] = None
    name: Optional[str] = None
    logs: Optional[str] = None
    params: Optional[Union[str, dict]] = None
    returns: Optional[Union[dict, 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 a ToolEvent to a Span"""
        span_attributes = {
            "tool.id": str(self.id),
            "tool.name": self.name,
            "session.id": str(self.session_id),
            "agent.id": str(self.agent_id),
            "gen_ai.tool.name": self.name,
        }

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

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

        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"tool:{self.name}",
            trace_id=trace_id,
            span_id=str(self.id),
            parent_span_id=parent_span_id,
            kind=SpanKind.INTERNAL,
            start_time=init_timestamp_ns,
            end_time=datetime_to_ns(self.end_timestamp),
            project_id=project_id,