Python: DevUI Fix Serialization, Timestamp and Other Issues (#1584)

* refactor(devui): adopt standard OpenAI lifecycle events for agents and workflows

- Replace custom workflow events with OpenAI Responses API standard lifecycle events
- Add AgentStartedEvent, AgentCompletedEvent, AgentFailedEvent for clean separation
- Implement ExecutorActionItem for workflow executor tracking
- Convert informational events to trace events to reduce noise
- Update README mapper table with comprehensive event mappings
- Maintain full backward compatibility with legacy events

* fix(devui): resolve timestamp overwriting and Content serialization errors

- Fix tool call timestamps being overwritten on each render (#1483)
- Add recursive Content serialization to handle ChatMessage and nested objects (#1548)
- Implement proper MCP tool cleanup on server shutdown
- Add timestamp field to function_result.complete events
- Enhance credential and client resource cleanup

Fixes #1483, #1548
Partial improvements for #1476
This commit is contained in:
Victor Dibia
2025-10-23 11:19:20 -07:00
committed by GitHub
Unverified
parent 064ee8afbe
commit 6b66a34609
21 changed files with 1859 additions and 682 deletions
@@ -127,7 +127,7 @@ class EntityDiscovery:
# Cache the loaded object
self._loaded_objects[entity_id] = entity_obj
logger.info(f"Successfully loaded entity: {entity_id} (type: {enriched_info.type})")
logger.info(f"Successfully loaded entity: {entity_id} (type: {enriched_info.type})")
return entity_obj
@@ -217,7 +217,7 @@ class EntityDiscovery:
if entity_info and "lazy_loaded" in entity_info.metadata:
entity_info.metadata["lazy_loaded"] = False
logger.info(f"♻️ Entity invalidated: {entity_id} (will reload on next access)")
logger.info(f"Entity invalidated: {entity_id} (will reload on next access)")
def invalidate_all(self) -> None:
"""Invalidate all cached entities.
@@ -217,6 +217,11 @@ class AgentFrameworkExecutor:
Agent update events and trace events
"""
try:
# Emit agent lifecycle start event
from .models._openai_custom import AgentStartedEvent
yield AgentStartedEvent()
# Convert input to proper ChatMessage or string
user_message = self._convert_input_to_chat_message(request.input)
@@ -266,8 +271,19 @@ class AgentFrameworkExecutor:
else:
raise ValueError("Agent must implement either run() or run_stream() method")
# Emit agent lifecycle completion event
from .models._openai_custom import AgentCompletedEvent
yield AgentCompletedEvent()
except Exception as e:
logger.error(f"Error in agent execution: {e}")
# Emit agent lifecycle failure event
from .models._openai_custom import AgentFailedEvent
yield AgentFailedEvent(error=e)
# Still yield the error for backward compatibility
yield {"type": "error", "message": f"Agent execution error: {e!s}"}
async def _execute_workflow(
@@ -284,14 +300,9 @@ class AgentFrameworkExecutor:
Workflow events and trace events
"""
try:
# Get input data - prefer structured data from extra_body
input_data: str | list[Any] | dict[str, Any]
if request.extra_body and isinstance(request.extra_body, dict) and request.extra_body.get("input_data"):
input_data = request.extra_body.get("input_data") # type: ignore
logger.debug(f"Using structured input_data from extra_body: {type(input_data)}")
else:
input_data = request.input
logger.debug(f"Using input field as fallback: {type(input_data)}")
# Get input data directly from request.input field
input_data = request.input
logger.debug(f"Using input field: {type(input_data)}")
# Parse input based on workflow's expected input type
parsed_input = await self._parse_workflow_input(workflow, input_data)
@@ -4,17 +4,32 @@
import json
import logging
import time
import uuid
from collections import OrderedDict
from collections.abc import Sequence
from datetime import datetime
from typing import Any, Union
from uuid import uuid4
from openai.types.responses import (
Response,
ResponseContentPartAddedEvent,
ResponseCreatedEvent,
ResponseError,
ResponseFailedEvent,
ResponseInProgressEvent,
)
from .models import (
AgentFrameworkRequest,
CustomResponseOutputItemAddedEvent,
CustomResponseOutputItemDoneEvent,
ExecutorActionItem,
InputTokensDetails,
OpenAIResponse,
OutputTokensDetails,
ResponseCompletedEvent,
ResponseErrorEvent,
ResponseFunctionCallArgumentsDeltaEvent,
ResponseFunctionResultComplete,
@@ -41,6 +56,56 @@ EventType = Union[
]
def _serialize_content_recursive(value: Any) -> Any:
"""Recursively serialize Agent Framework Content objects to JSON-compatible values.
This handles nested Content objects (like TextContent inside FunctionResultContent.result)
that can't be directly serialized by json.dumps().
Args:
value: Value to serialize (can be Content object, dict, list, primitive, etc.)
Returns:
JSON-serializable version with all Content objects converted to dicts/primitives
"""
# Handle None and basic JSON-serializable types
if value is None or isinstance(value, (str, int, float, bool)):
return value
# Check if it's a SerializationMixin (includes all Content types)
# Content objects have to_dict() method
if hasattr(value, "to_dict") and callable(getattr(value, "to_dict", None)):
try:
return value.to_dict()
except Exception as e:
# If to_dict() fails, fall through to other methods
logger.debug(f"Failed to serialize with to_dict(): {e}")
# Handle dictionaries - recursively process values
if isinstance(value, dict):
return {key: _serialize_content_recursive(val) for key, val in value.items()}
# Handle lists and tuples - recursively process elements
if isinstance(value, (list, tuple)):
serialized = [_serialize_content_recursive(item) for item in value]
# For single-item lists containing text Content, extract just the text
# This handles the MCP case where result = [TextContent(text="Hello")]
# and we want output = "Hello" not output = '[{"type": "text", "text": "Hello"}]'
if len(serialized) == 1 and isinstance(serialized[0], dict) and serialized[0].get("type") == "text":
return serialized[0].get("text", "")
return serialized
# For other objects with model_dump(), try that
if hasattr(value, "model_dump") and callable(getattr(value, "model_dump", None)):
try:
return value.model_dump()
except Exception as e:
logger.debug(f"Failed to serialize with model_dump(): {e}")
# Return as-is and let json.dumps handle it (may raise TypeError for non-serializable types)
return value
class MessageMapper:
"""Maps Agent Framework messages/responses to OpenAI format."""
@@ -102,6 +167,12 @@ class MessageMapper:
)
]
# Handle Agent lifecycle events first
from .models._openai_custom import AgentCompletedEvent, AgentFailedEvent, AgentStartedEvent
if isinstance(raw_event, (AgentStartedEvent, AgentCompletedEvent, AgentFailedEvent)):
return await self._convert_agent_lifecycle_event(raw_event, context)
# Import Agent Framework types for proper isinstance checks
try:
from agent_framework import AgentRunResponse, AgentRunResponseUpdate, WorkflowEvent
@@ -245,6 +316,7 @@ class MessageMapper:
"content_index": 0,
"output_index": 0,
"request_id": str(request_key), # For usage accumulation
"request": request, # Store the request for model name access
# Track active function calls: {call_id: {name, item_id, args_chunks}}
"active_function_calls": {},
}
@@ -267,7 +339,7 @@ class MessageMapper:
return int(context["sequence_counter"])
async def _convert_agent_update(self, update: Any, context: dict[str, Any]) -> Sequence[Any]:
"""Convert AgentRunResponseUpdate to OpenAI events using comprehensive content mapping.
"""Convert agent text updates to proper content part events.
Args:
update: Agent run response update
@@ -283,10 +355,60 @@ class MessageMapper:
if not hasattr(update, "contents") or not update.contents:
return events
# Check if we're streaming text content
has_text_content = any(content.__class__.__name__ == "TextContent" for content in update.contents)
# If we have text content and haven't created a message yet, create one
if has_text_content and "current_message_id" not in context:
message_id = f"msg_{uuid4().hex[:8]}"
context["current_message_id"] = message_id
context["output_index"] = context.get("output_index", -1) + 1
# Add message output item
events.append(
ResponseOutputItemAddedEvent(
type="response.output_item.added",
output_index=context["output_index"],
sequence_number=self._next_sequence(context),
item=ResponseOutputMessage(
type="message", id=message_id, role="assistant", content=[], status="in_progress"
),
)
)
# Add content part for text
context["content_index"] = 0
events.append(
ResponseContentPartAddedEvent(
type="response.content_part.added",
output_index=context["output_index"],
content_index=context["content_index"],
item_id=message_id,
sequence_number=self._next_sequence(context),
part=ResponseOutputText(type="output_text", text="", annotations=[]),
)
)
# Process each content item
for content in update.contents:
content_type = content.__class__.__name__
if content_type in self.content_mappers:
# Special handling for TextContent to use proper delta events
if content_type == "TextContent" and "current_message_id" in context:
# Stream text content via proper delta events
events.append(
ResponseTextDeltaEvent(
type="response.output_text.delta",
output_index=context["output_index"],
content_index=context.get("content_index", 0),
item_id=context["current_message_id"],
delta=content.text,
logprobs=[], # We don't have logprobs from Agent Framework
sequence_number=self._next_sequence(context),
)
)
elif content_type in self.content_mappers:
# Use existing mappers for other content types
mapped_events = await self.content_mappers[content_type](content, context)
if mapped_events is not None: # Handle None returns (e.g., UsageContent)
if isinstance(mapped_events, list):
@@ -297,7 +419,9 @@ class MessageMapper:
# Graceful fallback for unknown content types
events.append(await self._create_unknown_content_event(content, context))
context["content_index"] += 1
# Don't increment content_index for text deltas within the same part
if content_type != "TextContent":
context["content_index"] = context.get("content_index", 0) + 1
except Exception as e:
logger.warning(f"Error converting agent update: {e}")
@@ -358,8 +482,105 @@ class MessageMapper:
return events
async def _convert_agent_lifecycle_event(self, event: Any, context: dict[str, Any]) -> Sequence[Any]:
"""Convert agent lifecycle events to OpenAI response events.
Args:
event: AgentStartedEvent, AgentCompletedEvent, or AgentFailedEvent
context: Conversion context
Returns:
List of OpenAI response stream events
"""
from .models._openai_custom import AgentCompletedEvent, AgentFailedEvent, AgentStartedEvent
try:
# Get model name from context (the agent name)
model_name = context.get("request", {}).model if context.get("request") else "agent"
if isinstance(event, AgentStartedEvent):
execution_id = f"agent_{uuid4().hex[:12]}"
context["execution_id"] = execution_id
# Create Response object
response_obj = Response(
id=f"resp_{execution_id}",
object="response",
created_at=float(time.time()),
model=model_name,
output=[],
status="in_progress",
parallel_tool_calls=False,
tool_choice="none",
tools=[],
)
# Emit both created and in_progress events
return [
ResponseCreatedEvent(
type="response.created", sequence_number=self._next_sequence(context), response=response_obj
),
ResponseInProgressEvent(
type="response.in_progress", sequence_number=self._next_sequence(context), response=response_obj
),
]
if isinstance(event, AgentCompletedEvent):
execution_id = context.get("execution_id", f"agent_{uuid4().hex[:12]}")
response_obj = Response(
id=f"resp_{execution_id}",
object="response",
created_at=float(time.time()),
model=model_name,
output=[],
status="completed",
parallel_tool_calls=False,
tool_choice="none",
tools=[],
)
return [
ResponseCompletedEvent(
type="response.completed", sequence_number=self._next_sequence(context), response=response_obj
)
]
if isinstance(event, AgentFailedEvent):
execution_id = context.get("execution_id", f"agent_{uuid4().hex[:12]}")
# Create error object
response_error = ResponseError(
message=str(event.error) if event.error else "Unknown error", code="server_error"
)
response_obj = Response(
id=f"resp_{execution_id}",
object="response",
created_at=float(time.time()),
model=model_name,
output=[],
status="failed",
error=response_error,
parallel_tool_calls=False,
tool_choice="none",
tools=[],
)
return [
ResponseFailedEvent(
type="response.failed", sequence_number=self._next_sequence(context), response=response_obj
)
]
return []
except Exception as e:
logger.warning(f"Error converting agent lifecycle event: {e}")
return [await self._create_error_event(str(e), context)]
async def _convert_workflow_event(self, event: Any, context: dict[str, Any]) -> Sequence[Any]:
"""Convert workflow event to structured OpenAI events.
"""Convert workflow events to standard OpenAI event objects.
Args:
event: Workflow event
@@ -369,22 +590,247 @@ class MessageMapper:
List of OpenAI response stream events
"""
try:
event_class = event.__class__.__name__
# Response-level events - construct proper OpenAI objects
if event_class == "WorkflowStartedEvent":
workflow_id = getattr(event, "workflow_id", str(uuid4()))
context["workflow_id"] = workflow_id
# Import Response type for proper construction
from openai.types.responses import Response
# Return proper OpenAI event objects
events: list[Any] = []
# Determine the model name - use request model or default to "workflow"
# The request model will be the agent name for agents, workflow name for workflows
model_name = context.get("request", {}).model if context.get("request") else "workflow"
# Create a full Response object with all required fields
response_obj = Response(
id=f"resp_{workflow_id}",
object="response",
created_at=float(time.time()),
model=model_name, # Use the actual model/agent name
output=[], # Empty output list initially
status="in_progress",
# Required fields with safe defaults
parallel_tool_calls=False,
tool_choice="none",
tools=[],
)
# First emit response.created
events.append(
ResponseCreatedEvent(
type="response.created", sequence_number=self._next_sequence(context), response=response_obj
)
)
# Then emit response.in_progress (reuse same response object)
events.append(
ResponseInProgressEvent(
type="response.in_progress", sequence_number=self._next_sequence(context), response=response_obj
)
)
return events
if event_class in ["WorkflowCompletedEvent", "WorkflowOutputEvent"]:
workflow_id = context.get("workflow_id", str(uuid4()))
# Import Response type for proper construction
from openai.types.responses import Response
# Get model name from context
model_name = context.get("request", {}).model if context.get("request") else "workflow"
# Create a full Response object for completed state
response_obj = Response(
id=f"resp_{workflow_id}",
object="response",
created_at=float(time.time()),
model=model_name,
output=[], # Output should be populated by this point from text streaming
status="completed",
parallel_tool_calls=False,
tool_choice="none",
tools=[],
)
return [
ResponseCompletedEvent(
type="response.completed", sequence_number=self._next_sequence(context), response=response_obj
)
]
if event_class == "WorkflowFailedEvent":
workflow_id = context.get("workflow_id", str(uuid4()))
error_info = getattr(event, "error", None)
# Import Response and ResponseError types
from openai.types.responses import Response, ResponseError
# Get model name from context
model_name = context.get("request", {}).model if context.get("request") else "workflow"
# Create error object
error_message = str(error_info) if error_info else "Unknown error"
# Create ResponseError object (code must be one of the allowed values)
response_error = ResponseError(
message=error_message,
code="server_error", # Use generic server_error code for workflow failures
)
# Create a full Response object for failed state
response_obj = Response(
id=f"resp_{workflow_id}",
object="response",
created_at=float(time.time()),
model=model_name,
output=[],
status="failed",
error=response_error,
parallel_tool_calls=False,
tool_choice="none",
tools=[],
)
return [
ResponseFailedEvent(
type="response.failed", sequence_number=self._next_sequence(context), response=response_obj
)
]
# Executor-level events (output items)
if event_class == "ExecutorInvokedEvent":
executor_id = getattr(event, "executor_id", "unknown")
item_id = f"exec_{executor_id}_{uuid4().hex[:8]}"
context[f"exec_item_{executor_id}"] = item_id
context["output_index"] = context.get("output_index", -1) + 1
# Create ExecutorActionItem with proper type
executor_item = ExecutorActionItem(
type="executor_action",
id=item_id,
executor_id=executor_id,
status="in_progress",
metadata=getattr(event, "metadata", {}),
)
# Use our custom event type that accepts ExecutorActionItem
return [
CustomResponseOutputItemAddedEvent(
type="response.output_item.added",
output_index=context["output_index"],
sequence_number=self._next_sequence(context),
item=executor_item,
)
]
if event_class == "ExecutorCompletedEvent":
executor_id = getattr(event, "executor_id", "unknown")
item_id = context.get(f"exec_item_{executor_id}", f"exec_{executor_id}_unknown")
# Create ExecutorActionItem with completed status
# ExecutorCompletedEvent uses 'data' field, not 'result'
executor_item = ExecutorActionItem(
type="executor_action",
id=item_id,
executor_id=executor_id,
status="completed",
result=getattr(event, "data", None),
)
# Use our custom event type
return [
CustomResponseOutputItemDoneEvent(
type="response.output_item.done",
output_index=context.get("output_index", 0),
sequence_number=self._next_sequence(context),
item=executor_item,
)
]
if event_class == "ExecutorFailedEvent":
executor_id = getattr(event, "executor_id", "unknown")
item_id = context.get(f"exec_item_{executor_id}", f"exec_{executor_id}_unknown")
error_info = getattr(event, "error", None)
# Create ExecutorActionItem with failed status
executor_item = ExecutorActionItem(
type="executor_action",
id=item_id,
executor_id=executor_id,
status="failed",
error={"message": str(error_info)} if error_info else None,
)
# Use our custom event type
return [
CustomResponseOutputItemDoneEvent(
type="response.output_item.done",
output_index=context.get("output_index", 0),
sequence_number=self._next_sequence(context),
item=executor_item,
)
]
# Handle informational workflow events (status, warnings, errors)
if event_class in ["WorkflowStatusEvent", "WorkflowWarningEvent", "WorkflowErrorEvent", "RequestInfoEvent"]:
# These are informational events that don't map to OpenAI lifecycle events
# Convert them to trace events for debugging visibility
event_data: dict[str, Any] = {}
# Extract relevant data based on event type
if event_class == "WorkflowStatusEvent":
event_data["state"] = str(getattr(event, "state", "unknown"))
elif event_class == "WorkflowWarningEvent":
event_data["message"] = str(getattr(event, "message", ""))
elif event_class == "WorkflowErrorEvent":
event_data["message"] = str(getattr(event, "message", ""))
event_data["error"] = str(getattr(event, "error", ""))
elif event_class == "RequestInfoEvent":
request_info = getattr(event, "data", {})
event_data["request_info"] = request_info if isinstance(request_info, dict) else str(request_info)
# Create a trace event for debugging
trace_event = ResponseTraceEventComplete(
type="response.trace.complete",
data={
"trace_type": "workflow_info",
"event_type": event_class,
"data": event_data,
"timestamp": datetime.now().isoformat(),
},
span_id=f"workflow_info_{uuid4().hex[:8]}",
item_id=context["item_id"],
output_index=context.get("output_index", 0),
sequence_number=self._next_sequence(context),
)
return [trace_event]
# For unknown/legacy events, still emit as workflow event for backward compatibility
# Get event data and serialize if it's a SerializationMixin
event_data = getattr(event, "data", None)
if event_data is not None and hasattr(event_data, "to_dict"):
raw_event_data = getattr(event, "data", None)
serialized_event_data: dict[str, Any] | str | None = raw_event_data
if raw_event_data is not None and hasattr(raw_event_data, "to_dict"):
# SerializationMixin objects - convert to dict for JSON serialization
try:
event_data = event_data.to_dict()
serialized_event_data = raw_event_data.to_dict()
except Exception as e:
logger.debug(f"Failed to serialize event data with to_dict(): {e}")
event_data = str(event_data)
serialized_event_data = str(raw_event_data)
# Create structured workflow event
# Create structured workflow event (keeping for backward compatibility)
workflow_event = ResponseWorkflowEventComplete(
type="response.workflow_event.complete",
data={
"event_type": event.__class__.__name__,
"data": event_data,
"data": serialized_event_data,
"executor_id": getattr(event, "executor_id", None),
"timestamp": datetime.now().isoformat(),
},
@@ -394,6 +840,7 @@ class MessageMapper:
sequence_number=self._next_sequence(context),
)
logger.debug(f"Unhandled workflow event type: {event_class}, emitting as legacy workflow event")
return [workflow_event]
except Exception as e:
@@ -538,8 +985,16 @@ class MessageMapper:
result = getattr(content, "result", None)
exception = getattr(content, "exception", None)
# Convert result to string
output = result if isinstance(result, str) else json.dumps(result) if result is not None else ""
# Convert result to string, handling nested Content objects from MCP tools
if isinstance(result, str):
output = result
elif result is not None:
# Recursively serialize any nested Content objects (e.g., from MCP tools)
serialized = _serialize_content_recursive(result)
# Convert to JSON string if still not a string
output = serialized if isinstance(serialized, str) else json.dumps(serialized)
else:
output = ""
# Determine status based on exception
status = "incomplete" if exception else "completed"
@@ -556,6 +1011,7 @@ class MessageMapper:
item_id=item_id,
output_index=context["output_index"],
sequence_number=self._next_sequence(context),
timestamp=datetime.now().isoformat(),
)
async def _map_error_content(self, content: Any, context: dict[str, Any]) -> ResponseErrorEvent:
@@ -723,7 +1179,7 @@ class MessageMapper:
async def _create_unknown_content_event(self, content: Any, context: dict[str, Any]) -> ResponseStreamEvent:
"""Create event for unknown content types."""
content_type = content.__class__.__name__
text = f"⚠️ Unknown content type: {content_type}\n"
text = f"Warning: Unknown content type: {content_type}\n"
return self._create_text_delta_event(text, context)
async def _create_error_response(self, error_message: str, request: AgentFrameworkRequest) -> OpenAIResponse:
@@ -85,19 +85,25 @@ class DevServer:
return self.executor
async def _cleanup_entities(self) -> None:
"""Cleanup entity resources (close clients, credentials, etc.)."""
"""Cleanup entity resources (close clients, MCP tools, credentials, etc.)."""
if not self.executor:
return
logger.info("Cleaning up entity resources...")
entities = self.executor.entity_discovery.list_entities()
closed_count = 0
mcp_tools_closed = 0
credentials_closed = 0
for entity_info in entities:
try:
entity_obj = self.executor.entity_discovery.get_entity_object(entity_info.id)
# Close chat clients and their credentials
if entity_obj and hasattr(entity_obj, "chat_client"):
client = entity_obj.chat_client
# Close the chat client itself
if hasattr(client, "close") and callable(client.close):
if inspect.iscoroutinefunction(client.close):
await client.close()
@@ -105,11 +111,47 @@ class DevServer:
client.close()
closed_count += 1
logger.debug(f"Closed client for entity: {entity_info.id}")
# Close credentials attached to chat clients (e.g., AzureCliCredential)
credential_attrs = ["credential", "async_credential", "_credential", "_async_credential"]
for attr in credential_attrs:
if hasattr(client, attr):
cred = getattr(client, attr)
if cred and hasattr(cred, "close") and callable(cred.close):
try:
if inspect.iscoroutinefunction(cred.close):
await cred.close()
else:
cred.close()
credentials_closed += 1
logger.debug(f"Closed credential for entity: {entity_info.id}")
except Exception as e:
logger.warning(f"Error closing credential for {entity_info.id}: {e}")
# Close MCP tools (framework tracks them in _local_mcp_tools)
if entity_obj and hasattr(entity_obj, "_local_mcp_tools"):
for mcp_tool in entity_obj._local_mcp_tools:
if hasattr(mcp_tool, "close") and callable(mcp_tool.close):
try:
if inspect.iscoroutinefunction(mcp_tool.close):
await mcp_tool.close()
else:
mcp_tool.close()
mcp_tools_closed += 1
tool_name = getattr(mcp_tool, "name", "unknown")
logger.debug(f"Closed MCP tool '{tool_name}' for entity: {entity_info.id}")
except Exception as e:
logger.warning(f"Error closing MCP tool for {entity_info.id}: {e}")
except Exception as e:
logger.warning(f"Error closing entity {entity_info.id}: {e}")
if closed_count > 0:
logger.info(f"Closed {closed_count} entity client(s)")
if credentials_closed > 0:
logger.info(f"Closed {credentials_closed} credential(s)")
if mcp_tools_closed > 0:
logger.info(f"Closed {mcp_tools_closed} MCP tool(s)")
def create_app(self) -> FastAPI:
"""Create the FastAPI application."""
@@ -30,6 +30,9 @@ from openai.types.shared import Metadata, ResponsesModel
from ._discovery_models import DiscoveryResponse, EntityInfo
from ._openai_custom import (
AgentFrameworkRequest,
CustomResponseOutputItemAddedEvent,
CustomResponseOutputItemDoneEvent,
ExecutorActionItem,
OpenAIError,
ResponseFunctionResultComplete,
ResponseTraceEvent,
@@ -46,8 +49,11 @@ __all__ = [
"Conversation",
"ConversationDeletedResource",
"ConversationItem",
"CustomResponseOutputItemAddedEvent",
"CustomResponseOutputItemDoneEvent",
"DiscoveryResponse",
"EntityInfo",
"ExecutorActionItem",
"InputTokensDetails",
"Metadata",
"OpenAIError",
@@ -8,6 +8,7 @@ to support Agent Framework specific features like workflows and traces.
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Literal
from pydantic import BaseModel, ConfigDict
@@ -15,6 +16,69 @@ from pydantic import BaseModel, ConfigDict
# Custom Agent Framework OpenAI event types for structured data
# Agent lifecycle events - simple and clear
class AgentStartedEvent:
"""Event emitted when an agent starts execution."""
pass
class AgentCompletedEvent:
"""Event emitted when an agent completes execution successfully."""
pass
@dataclass
class AgentFailedEvent:
"""Event emitted when an agent fails during execution."""
error: Exception | None = None
class ExecutorActionItem(BaseModel):
"""Custom item type for workflow executor actions.
This is a DevUI-specific extension to represent workflow executors as output items.
Since OpenAI's ResponseOutputItemAddedEvent only accepts specific item types,
and executor actions are not part of the standard, we need this custom type.
"""
type: Literal["executor_action"] = "executor_action"
id: str
executor_id: str
status: Literal["in_progress", "completed", "failed", "cancelled"] = "in_progress"
metadata: dict[str, Any] | None = None
result: Any | None = None
error: dict[str, Any] | None = None
class CustomResponseOutputItemAddedEvent(BaseModel):
"""Custom version of ResponseOutputItemAddedEvent that accepts any item type.
This allows us to emit executor action items while maintaining the same
event structure as OpenAI's standard.
"""
type: Literal["response.output_item.added"] = "response.output_item.added"
output_index: int
sequence_number: int
item: dict[str, Any] | ExecutorActionItem | Any # Flexible item type
class CustomResponseOutputItemDoneEvent(BaseModel):
"""Custom version of ResponseOutputItemDoneEvent that accepts any item type.
This allows us to emit executor action items while maintaining the same
event structure as OpenAI's standard.
"""
type: Literal["response.output_item.done"] = "response.output_item.done"
output_index: int
sequence_number: int
item: dict[str, Any] | ExecutorActionItem | Any # Flexible item type
class ResponseWorkflowEventComplete(BaseModel):
"""Complete workflow event data."""
@@ -57,6 +121,7 @@ class ResponseFunctionResultComplete(BaseModel):
item_id: str
output_index: int = 0
sequence_number: int
timestamp: str | None = None # Optional timestamp for UI display
# Agent Framework extension fields
@@ -64,7 +129,7 @@ class AgentFrameworkExtraBody(BaseModel):
"""Agent Framework specific routing fields for OpenAI requests."""
entity_id: str
input_data: dict[str, Any] | None = None
# input_data removed - now using standard input field for all data
model_config = ConfigDict(extra="allow")
@@ -80,7 +145,7 @@ class AgentFrameworkRequest(BaseModel):
# All OpenAI fields from ResponseCreateParams
model: str # Used as entity_id in DevUI!
input: str | list[Any] # ResponseInputParam
input: str | list[Any] | dict[str, Any] # ResponseInputParam + dict for workflow structured input
stream: bool | None = False
# OpenAI conversation parameter (standard!)
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
@@ -5,7 +5,7 @@
<link rel="icon" type="image/svg+xml" href="/agentframework.svg" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>Agent Framework Dev UI</title>
<script type="module" crossorigin src="/assets/index-DmL7WSFa.js"></script>
<script type="module" crossorigin src="/assets/index-D_Y1oSGu.js"></script>
<link rel="stylesheet" crossorigin href="/assets/index-CE4pGoXh.css">
</head>
<body>