Dynamic UI Tools

import asyncio
import json
import logging

from meshagent.api.messaging import JsonContent, TextContent
from meshagent.api.room_server_client import RoomClient
from meshagent.agents import SingleRoomAgent
from meshagent.agents.llmrunner import LLMTaskRunner
from meshagent.otel import otel_config
from meshagent.openai import OpenAIResponsesAdapter
from meshagent.tools import LocalRoomTool, ToolContext, Toolkit

otel_config(service_name="my-service")
log = logging.getLogger("my-service")


async def save_to_storage(room: RoomClient, path: str, data: bytes):
    await room.storage.upload(path=path, data=data)


class Survey(LocalRoomTool):
    def __init__(self, *, room: RoomClient):
        super().__init__(
            room=room,
            name="survey",
            title="survey",
            description="a tool that conducts a survey of the participants",
            input_schema={
                "type": "object",
                "additionalProperties": False,
                "required": ["subject", "description", "name"],
                "properties": {
                    "subject": {
                        "type": "string",
                        "description": "The subject of the form",
                    },
                    "description": {
                        "type": "string",
                        "description": "The content to fill in (e.g. feedback, poll result)",
                    },
                    "name": {
                        "type": "string",
                        "description": "A short name to be used on the form",
                    },
                },
            },
        )

    async def execute(
        self, context: ToolContext, subject: str, description: str, name: str
    ):
        room = self.room
        participants = await self._wait_for_user_participants(room)
        log.info("Starting survey for %d participant(s)", len(participants))
        if not participants:
            return TextContent(
                text=(
                    "No user participants are available to survey. Open a MeshAgent UI "
                    "client in this room, confirm it is registered as a user participant, "
                    "and invoke the survey again."
                )
            )

        MAX_ATTEMPTS = 2

        async def ask_participant(p):
            errors = []
            for attempt in range(1, MAX_ATTEMPTS + 1):
                try:
                    log.info("→ ask_user attempt %d --> %s", attempt, p.id)
                    resp = await room.agents.invoke_tool(
                        toolkit="ui",
                        tool="ask_user",
                        participant_id=p.id,
                        input={
                            "subject": subject,
                            "help": description,
                            "form": [
                                {
                                    "input": {
                                        "multiline": False,
                                        "name": name,
                                        "description": description,
                                        "default_value": "",
                                    },
                                },
                            ],
                        },
                    )
                    answer = resp.json.get(name)
                    if answer:
                        log.info("participant_id", p.id, "response", answer)
                        return {"participant_id": p.id, "response": answer}
                    raise RuntimeError("empty or timed-out response")

                except Exception as exc:
                    errors.append(f"attempt {attempt}: {exc}")
                    if attempt < MAX_ATTEMPTS:
                        log.info("Retrying %s after: %s", p.id, exc)
                        await asyncio.sleep(1)  # brief back-off

            # All attempts failed – return aggregated error list
            return {"participant_id": p.id, "errors": errors}

        log.info("Surveying participants")
        tasks = [asyncio.create_task(ask_participant(p)) for p in participants]
        results = await asyncio.gather(*tasks)

        summary = {
            "meta": {  # save the prompt generated for the survey
                "subject": subject,
                "description": description,
                "name": name,
            },
            "success": {},
            "failed": {},
        }
        for item in results:
            pid = item["participant_id"]
            if "response" in item:
                summary["success"][pid] = item["response"]
            else:
                summary["failed"][pid] = item["errors"]

        # write survey results to the room
        log.info("Survey completed, writing raw results to Room storage")
        await save_to_storage(
            room=room,
            path=f"survey/{room.room_name}-{name}.json",
            data=json.dumps({"summary": summary}, indent=2).encode("utf-8"),
        )

        # summarize results
        log.info("Summarizing survey results")
        summary_schema = {
            "type": "object",
            "properties": {"summary": {"type": "string"}},
            "required": ["summary"],
            "additionalProperties": False,
        }
        runner = LLMTaskRunner(
            llm_adapter=OpenAIResponsesAdapter(),
            output_schema=summary_schema,
        )
        summary_resp = await runner.run(
            room=room,
            arguments={
                "prompt": f"Summarize these survey results:\n{json.dumps(summary)}",
                "model": None,
            },
            caller=context.caller,
        )
        if isinstance(summary_resp, JsonContent):
            summary_text = summary_resp.json.get("summary") or summary_resp.json.get(
                "result", ""
            )
        elif isinstance(summary_resp, TextContent):
            summary_text = summary_resp.text
        else:
            summary_text = str(summary_resp)
        log.info("Saving survey result summary")
        await save_to_storage(
            room=room,
            path=f"survey/{room.room_name}-{name}-summary.doc",
            data=summary_text.encode("utf-8"),
        )

        return TextContent(text=summary_text)

    async def _wait_for_user_participants(self, room: RoomClient, timeout: float = 5.0):
        deadline = asyncio.get_running_loop().time() + timeout
        while True:
            participants = [
                p for p in room.messaging.remote_participants if p.role == "user"
            ]
            if participants:
                return participants
            if asyncio.get_running_loop().time() >= deadline:
                return []
            await asyncio.sleep(0.25)


class SurveyToolkit(Toolkit):
    def __init__(self, *, room: RoomClient):
        super().__init__(
            name="survey-toolkit",
            title="survey-toolkit",
            description="a toolkit for conducting a survey",
            tools=[Survey(room=room)],
        )


class SurveyAgent(SingleRoomAgent):
    async def start(self, *, room: RoomClient) -> None:
        await room.messaging.enable()
        await super().start(room=room)

    async def get_exposed_toolkits(self) -> list[Toolkit]:
        return [SurveyToolkit(room=self.room)]


async def main() -> None:
    agent = SurveyAgent(title="survey-toolkit-host")
    await agent.run()


if __name__ == "__main__":
    asyncio.run(main())