From 3f5375339866b375896e9ba36941dea475e64e5c Mon Sep 17 00:00:00 2001 From: MFA-X-AI Date: Thu, 16 Jul 2026 02:27:05 +0800 Subject: [PATCH 1/2] feat(core): add structured response events to ResponseProtocol Adds a typed event union (tool_call, tool_result, usage) to the response models, a respond_event method on ResponseProtocol, turn-scoped event accumulation in ResponseHandler, and emission in SimpleToolOrchestrator around LLM usage and tool execution. UsageEvent subclasses LLMUsage so future usage fields flow through; tool results in events get the same json.dumps(default=repr) coercion as the conversation path. Duck-typed response handlers in the OpenAI/OpenAI-Responses/LiveKit adapters get a no-op respond_event. This gives adapters a structured surface for agent internals instead of final text only, as a prerequisite for an AG-UI adapter. --- src/xaibo/core/models/response.py | 43 ++++- src/xaibo/core/protocols/response.py | 14 +- src/xaibo/integrations/livekit/llm.py | 4 + .../orchestrator/simple_tool_orchestrator.py | 58 ++++-- src/xaibo/primitives/modules/response.py | 17 +- src/xaibo/server/adapters/openai.py | 4 + src/xaibo/server/adapters/openai_responses.py | 4 + tests/agents/test_response_events.py | 172 ++++++++++++++++++ 8 files changed, 291 insertions(+), 25 deletions(-) create mode 100644 tests/agents/test_response_events.py diff --git a/src/xaibo/core/models/response.py b/src/xaibo/core/models/response.py index 573e577..8079221 100644 --- a/src/xaibo/core/models/response.py +++ b/src/xaibo/core/models/response.py @@ -1,5 +1,9 @@ from enum import Enum -from typing import BinaryIO, Optional, List +from typing import BinaryIO, Optional, List, Any, Dict, Literal, Union + +from pydantic import BaseModel + +from xaibo.core.models.llm import LLMUsage class FileType(Enum): @@ -19,11 +23,44 @@ def __init__(self, content: BinaryIO, type: FileType) -> None: self.type = type +class ToolCallEvent(BaseModel): + """A tool is about to be executed""" + type: Literal["tool_call"] = "tool_call" + id: str + name: str + arguments: Dict[str, Any] + + +class ToolResultEvent(BaseModel): + """A tool execution finished""" + type: Literal["tool_result"] = "tool_result" + id: str + name: str + success: bool + result: Optional[Any] = None + error: Optional[str] = None + + +class UsageEvent(LLMUsage): + """Token usage reported by an LLM call""" + type: Literal["usage"] = "usage" + + +ResponseEvent = Union[ + ToolCallEvent, + ToolResultEvent, + UsageEvent, +] + + class Response: - """Model for responses that can include text and file attachments""" + """Model for responses that can include text, file attachments and structured events""" text: Optional[str] = None attachments: List[FileAttachment] = [] + events: List[ResponseEvent] - def __init__(self, text: Optional[str] = None, attachments: Optional[List[FileAttachment]] = None) -> None: + def __init__(self, text: Optional[str] = None, attachments: Optional[List[FileAttachment]] = None, + events: Optional[List[ResponseEvent]] = None) -> None: self.text = text self.attachments = attachments or [] + self.events = events or [] diff --git a/src/xaibo/core/protocols/response.py b/src/xaibo/core/protocols/response.py index 21eabff..a303b78 100644 --- a/src/xaibo/core/protocols/response.py +++ b/src/xaibo/core/protocols/response.py @@ -1,6 +1,6 @@ from typing import Protocol, BinaryIO, runtime_checkable -from xaibo.core.models.response import Response +from xaibo.core.models.response import Response, ResponseEvent @runtime_checkable @@ -53,4 +53,16 @@ async def respond(self, response: Response) -> None: Args: response: Response object containing text and attachments """ + ... + + async def respond_event(self, event: ResponseEvent) -> None: + """Report a structured event produced while generating the response. + + Events carry the internals of response generation (tool calls, tool + results, usage) so that adapters can surface them instead of only + the final text. + + Args: + event: The event to report + """ ... \ No newline at end of file diff --git a/src/xaibo/integrations/livekit/llm.py b/src/xaibo/integrations/livekit/llm.py index 3cb4200..e474b57 100644 --- a/src/xaibo/integrations/livekit/llm.py +++ b/src/xaibo/integrations/livekit/llm.py @@ -294,6 +294,10 @@ async def respond_text(self, text: str) -> None: """Handle streaming text chunks from the agent""" await chunk_queue.put(text) + async def respond_event(self, event) -> None: + """Agent internals are not forwarded to the voice pipeline""" + pass + async def get_response(self): """Return None as we handle streaming via respond_text""" return None diff --git a/src/xaibo/primitives/modules/orchestrator/simple_tool_orchestrator.py b/src/xaibo/primitives/modules/orchestrator/simple_tool_orchestrator.py index bf0fd1c..1c8c8b2 100644 --- a/src/xaibo/primitives/modules/orchestrator/simple_tool_orchestrator.py +++ b/src/xaibo/primitives/modules/orchestrator/simple_tool_orchestrator.py @@ -1,9 +1,15 @@ from xaibo.core.protocols import TextMessageHandlerProtocol, ResponseProtocol, LLMProtocol, ToolProviderProtocol, \ ConversationHistoryProtocol from xaibo.core.models.llm import LLMMessage, LLMOptions, LLMRole, LLMFunctionResult, LLMMessageContentType, LLMMessageContent +from xaibo.core.models.response import ToolCallEvent, ToolResultEvent, UsageEvent import json + +def _json_safe(value): + """Coerce a tool result into JSON-serializable form""" + return json.loads(json.dumps(value, default=repr)) + class SimpleToolOrchestrator(TextMessageHandlerProtocol): """ A text message handler that uses tools with increasing stress levels on failures. @@ -91,7 +97,10 @@ async def handle_text(self, text: str) -> None: ) llm_response = await self.llm.generate(conversation, options) - + + if llm_response.usage: + await self.response.respond_event(UsageEvent(**llm_response.usage.model_dump())) + # Add assistant response to conversation assistant_message = LLMMessage( role=LLMRole.FUNCTION if llm_response.tool_calls else LLMRole.ASSISTANT, @@ -111,33 +120,46 @@ async def handle_text(self, text: str) -> None: for tool_call in llm_response.tool_calls: tool_name = tool_call.name tool_args = tool_call.arguments - + + await self.response.respond_event(ToolCallEvent( + id=tool_call.id, + name=tool_name, + arguments=tool_args + )) + try: tool_result = await self.tool_provider.execute_tool(tool_name, tool_args) - - if tool_result.success: - tool_results.append({ - "id": tool_call.id, - "name": tool_name, - "result": json.dumps(tool_result.result, default=repr) - }) - else: - # Tool execution failed - tool_results.append({ - "id": tool_call.id, - "name": tool_name, - "error": f"Error: {tool_result.error}" - }) - stress_level += 0.1 + success = tool_result.success + result = tool_result.result if success else None + error = tool_result.error if not success else None except Exception as e: # Handle any exceptions during tool execution + success = False + result = None + error = str(e) + + if success: tool_results.append({ "id": tool_call.id, "name": tool_name, - "error": f"Error: {str(e)}" + "result": json.dumps(result, default=repr) + }) + else: + tool_results.append({ + "id": tool_call.id, + "name": tool_name, + "error": f"Error: {error}" }) stress_level += 0.1 + await self.response.respond_event(ToolResultEvent( + id=tool_call.id, + name=tool_name, + success=success, + result=_json_safe(result), + error=error + )) + conversation.append(LLMMessage( role=LLMRole.FUNCTION, tool_results=[ diff --git a/src/xaibo/primitives/modules/response.py b/src/xaibo/primitives/modules/response.py index 8d45a4b..0166317 100644 --- a/src/xaibo/primitives/modules/response.py +++ b/src/xaibo/primitives/modules/response.py @@ -1,7 +1,7 @@ from typing import BinaryIO from xaibo.core.protocols import ResponseProtocol -from xaibo.core.models import FileAttachment, FileType, Response +from xaibo.core.models import FileAttachment, FileType, Response, ResponseEvent class ResponseHandler(ResponseProtocol): @@ -9,7 +9,14 @@ def __init__(self, config: dict = None): self._response = Response() async def get_response(self) -> Response: - return self._response + # Events are turn-scoped: hand them off and reset + response = Response( + text=self._response.text, + attachments=self._response.attachments, + events=self._response.events + ) + self._response.events = [] + return response async def respond_text(self, response: str) -> None: if self._response.text is None: @@ -32,4 +39,8 @@ async def respond(self, response: Response) -> None: self._response.text = response.text else: self._response.text += response.text - self._response.attachments.extend(response.attachments) \ No newline at end of file + self._response.attachments.extend(response.attachments) + self._response.events.extend(response.events) + + async def respond_event(self, event: ResponseEvent) -> None: + self._response.events.append(event) \ No newline at end of file diff --git a/src/xaibo/server/adapters/openai.py b/src/xaibo/server/adapters/openai.py index 9b8d6f8..842c32e 100644 --- a/src/xaibo/server/adapters/openai.py +++ b/src/xaibo/server/adapters/openai.py @@ -127,6 +127,10 @@ async def respond_text(self, text: str) -> None: response = create_chunk_response({"content": text}) await queue.put(f"data: {json.dumps(response)}\n\n") + async def respond_event(self, event) -> None: + # The OpenAI chat completion grammar has no lane for agent internals + pass + async def get_response(self): return None diff --git a/src/xaibo/server/adapters/openai_responses.py b/src/xaibo/server/adapters/openai_responses.py index 3de8b56..f29aefa 100644 --- a/src/xaibo/server/adapters/openai_responses.py +++ b/src/xaibo/server/adapters/openai_responses.py @@ -412,6 +412,10 @@ async def respond_text(self, text: str) -> None: await queue.put(f"data: {json.dumps(event_data)}\n\n") self.accumulated_text += text + async def respond_event(self, event) -> None: + # Agent internals are not part of the responses API surface yet + pass + async def get_response(self): return None diff --git a/tests/agents/test_response_events.py b/tests/agents/test_response_events.py new file mode 100644 index 0000000..22ac60e --- /dev/null +++ b/tests/agents/test_response_events.py @@ -0,0 +1,172 @@ +import pytest + +from xaibo import AgentConfig, Xaibo, ConfigOverrides, ExchangeConfig +from xaibo.core.config import ModuleConfig +from xaibo.core.models.response import ( + Response, + ToolCallEvent, + ToolResultEvent, + UsageEvent, +) +from xaibo.core.models.tools import Tool, ToolResult +from xaibo.primitives.modules.conversation import SimpleConversation +from xaibo.primitives.modules.response import ResponseHandler + + +class StubToolProvider: + def __init__(self, result: ToolResult): + self.result = result + self.calls = [] + + async def list_tools(self): + return [Tool(name="get_time", description="Gets the current time")] + + async def execute_tool(self, tool_name, parameters): + self.calls.append((tool_name, parameters)) + return self.result + + +def make_agent(tool_provider, responses): + config = AgentConfig( + id="event-emitter", + modules=[ + ModuleConfig( + module="xaibo.primitives.modules.llm.MockLLM", + id="llm", + config={"responses": responses}, + ), + ModuleConfig( + module="xaibo.primitives.modules.orchestrator.SimpleToolOrchestrator", + id="orchestrator", + config={"max_thoughts": 5}, + ), + ], + ) + xaibo = Xaibo() + xaibo.register_agent(config) + return xaibo.get_agent_with("event-emitter", ConfigOverrides( + instances={ + "tools": tool_provider, + "history": SimpleConversation(), + }, + exchange=[ + ExchangeConfig(protocol="ToolProviderProtocol", provider="tools"), + ExchangeConfig(protocol="ConversationHistoryProtocol", provider="history"), + ], + )) + + +TOOL_CALL_RESPONSE = { + "content": "", + "tool_calls": [{"id": "call_1", "name": "get_time", "arguments": {"timezone": "UTC"}}], + "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15}, +} +FINAL_RESPONSE = { + "content": "It is noon.", + "usage": {"prompt_tokens": 20, "completion_tokens": 4, "total_tokens": 24}, +} + + +@pytest.mark.asyncio +async def test_response_handler_accumulates_events(): + handler = ResponseHandler() + + await handler.respond_event(ToolCallEvent(id="call_1", name="get_time", arguments={})) + await handler.respond_event(ToolResultEvent(id="call_1", name="get_time", success=True, result="12:00")) + await handler.respond_text("It is noon.") + + response = await handler.get_response() + assert response.text == "It is noon." + assert [e.type for e in response.events] == ["tool_call", "tool_result"] + + +@pytest.mark.asyncio +async def test_response_handler_merges_events_from_respond(): + handler = ResponseHandler() + + await handler.respond(Response(text="done", events=[UsageEvent(prompt_tokens=1, completion_tokens=2, total_tokens=3)])) + + response = await handler.get_response() + assert [e.type for e in response.events] == ["usage"] + + +@pytest.mark.asyncio +async def test_response_handler_does_not_replay_events_across_turns(): + handler = ResponseHandler() + + await handler.respond_event(ToolCallEvent(id="call_1", name="get_time", arguments={})) + first = await handler.get_response() + assert [e.type for e in first.events] == ["tool_call"] + + await handler.respond_event(UsageEvent(prompt_tokens=1, completion_tokens=2, total_tokens=3)) + second = await handler.get_response() + # Only the second turn's events, and the first turn's response is untouched + assert [e.type for e in second.events] == ["usage"] + assert [e.type for e in first.events] == ["tool_call"] + + +@pytest.mark.asyncio +async def test_orchestrator_emits_tool_and_usage_events(): + tools = StubToolProvider(ToolResult(success=True, result="12:00")) + agent = make_agent(tools, [TOOL_CALL_RESPONSE, FINAL_RESPONSE]) + + response = await agent.handle_text("What time is it?") + + assert response.text == "It is noon." + assert [e.type for e in response.events] == ["usage", "tool_call", "tool_result", "usage"] + + tool_call = response.events[1] + assert tool_call.id == "call_1" + assert tool_call.name == "get_time" + assert tool_call.arguments == {"timezone": "UTC"} + + tool_result = response.events[2] + assert tool_result.id == "call_1" + assert tool_result.success is True + assert tool_result.result == "12:00" + + assert response.events[0].total_tokens == 15 + assert response.events[3].total_tokens == 24 + + +@pytest.mark.asyncio +async def test_orchestrator_emits_failed_tool_result(): + tools = StubToolProvider(ToolResult(success=False, error="tool unavailable")) + agent = make_agent(tools, [TOOL_CALL_RESPONSE, FINAL_RESPONSE]) + + response = await agent.handle_text("What time is it?") + + tool_result = next(e for e in response.events if e.type == "tool_result") + assert tool_result.success is False + assert tool_result.error == "tool unavailable" + + +@pytest.mark.asyncio +async def test_orchestrator_emits_tool_result_on_exception(): + class ExplodingToolProvider(StubToolProvider): + async def execute_tool(self, tool_name, parameters): + raise RuntimeError("kaboom") + + tools = ExplodingToolProvider(None) + agent = make_agent(tools, [TOOL_CALL_RESPONSE, FINAL_RESPONSE]) + + response = await agent.handle_text("What time is it?") + + tool_result = next(e for e in response.events if e.type == "tool_result") + assert tool_result.success is False + assert "kaboom" in tool_result.error + + +@pytest.mark.asyncio +async def test_tool_result_event_is_json_serializable(): + from datetime import datetime + + tools = StubToolProvider(ToolResult(success=True, result={"at": datetime(2026, 7, 16, 12, 0)})) + agent = make_agent(tools, [TOOL_CALL_RESPONSE, FINAL_RESPONSE]) + + response = await agent.handle_text("What time is it?") + + tool_result = next(e for e in response.events if e.type == "tool_result") + # Non-JSON-serializable values are coerced (via repr), so events can go on the wire + tool_result.model_dump_json() + assert "2026" in tool_result.result["at"] From a6cd48d4c8510f4c4171dc8b6dfc1c8a35f3fe49 Mon Sep 17 00:00:00 2001 From: MFA-X-AI Date: Thu, 16 Jul 2026 17:12:48 +0800 Subject: [PATCH 2/2] feat(core): make respond_event an optional response handler capability MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Response handlers in xaibo support what they support — partial implementations are the norm and nothing enforces the full protocol. Per standup discussion, respond_event follows the same rule rather than becoming a hard requirement: SimpleToolOrchestrator emits through an _emit_event guard that checks the handler and skips when the method is missing, so response handlers that predate it keep working unchanged. Handlers that want events extend to support them. A skipped emission never crosses the observability proxy and therefore cannot appear in the debug event trace, so the guard logs a debug line instead. The in-repo adapter handlers keep their no-op respond_event deliberately: an implemented-but-ignored event still crosses the proxy, keeping tool calls and usage visible in the debug trace when serving through these adapters. Adds a regression test running a legacy two-method response handler through the event-emitting orchestrator. --- .../orchestrator/simple_tool_orchestrator.py | 17 +++++-- tests/agents/test_response_events.py | 49 +++++++++++++++++++ 2 files changed, 63 insertions(+), 3 deletions(-) diff --git a/src/xaibo/primitives/modules/orchestrator/simple_tool_orchestrator.py b/src/xaibo/primitives/modules/orchestrator/simple_tool_orchestrator.py index 1c8c8b2..faec988 100644 --- a/src/xaibo/primitives/modules/orchestrator/simple_tool_orchestrator.py +++ b/src/xaibo/primitives/modules/orchestrator/simple_tool_orchestrator.py @@ -4,6 +4,9 @@ from xaibo.core.models.response import ToolCallEvent, ToolResultEvent, UsageEvent import json +import logging + +logger = logging.getLogger(__name__) def _json_safe(value): @@ -55,6 +58,14 @@ def __init__(self, self.tool_provider: ToolProviderProtocol = tool_provider self.history = history + async def _emit_event(self, event) -> None: + """Send a response event if the response handler supports it.""" + respond_event = getattr(self.response, "respond_event", None) + if respond_event is None: + logger.debug("Response handler does not implement respond_event; dropping %s event", event.type) + return + await respond_event(event) + async def handle_text(self, text: str) -> None: """ Process a user text message, potentially using tools to generate a response. @@ -99,7 +110,7 @@ async def handle_text(self, text: str) -> None: llm_response = await self.llm.generate(conversation, options) if llm_response.usage: - await self.response.respond_event(UsageEvent(**llm_response.usage.model_dump())) + await self._emit_event(UsageEvent(**llm_response.usage.model_dump())) # Add assistant response to conversation assistant_message = LLMMessage( @@ -121,7 +132,7 @@ async def handle_text(self, text: str) -> None: tool_name = tool_call.name tool_args = tool_call.arguments - await self.response.respond_event(ToolCallEvent( + await self._emit_event(ToolCallEvent( id=tool_call.id, name=tool_name, arguments=tool_args @@ -152,7 +163,7 @@ async def handle_text(self, text: str) -> None: }) stress_level += 0.1 - await self.response.respond_event(ToolResultEvent( + await self._emit_event(ToolResultEvent( id=tool_call.id, name=tool_name, success=success, diff --git a/tests/agents/test_response_events.py b/tests/agents/test_response_events.py index 22ac60e..e5f308c 100644 --- a/tests/agents/test_response_events.py +++ b/tests/agents/test_response_events.py @@ -157,6 +157,55 @@ async def execute_tool(self, tool_name, parameters): assert "kaboom" in tool_result.error +@pytest.mark.asyncio +async def test_legacy_response_handler_without_respond_event_still_works(): + """A handler that predates respond_event is skipped, not crashed (events are an optional capability).""" + class LegacyResponseHandler: + def __init__(self): + self._text = None + + async def respond_text(self, text): + self._text = text + + async def get_response(self): + return Response(text=self._text) + + tools = StubToolProvider(ToolResult(success=True, result="12:00")) + config = AgentConfig( + id="event-emitter", + modules=[ + ModuleConfig( + module="xaibo.primitives.modules.llm.MockLLM", + id="llm", + config={"responses": [TOOL_CALL_RESPONSE, FINAL_RESPONSE]}, + ), + ModuleConfig( + module="xaibo.primitives.modules.orchestrator.SimpleToolOrchestrator", + id="orchestrator", + config={"max_thoughts": 5}, + ), + ], + ) + xaibo = Xaibo() + xaibo.register_agent(config) + agent = xaibo.get_agent_with("event-emitter", ConfigOverrides( + instances={ + "tools": tools, + "history": SimpleConversation(), + "__response__": LegacyResponseHandler(), + }, + exchange=[ + ExchangeConfig(protocol="ToolProviderProtocol", provider="tools"), + ExchangeConfig(protocol="ConversationHistoryProtocol", provider="history"), + ], + )) + + response = await agent.handle_text("What time is it?") + + assert response.text == "It is noon." + assert len(tools.calls) == 1 + + @pytest.mark.asyncio async def test_tool_result_event_is_json_serializable(): from datetime import datetime