Function _run_tool

tests/test_esc_reject_message_dispatch.py:46–57

(ctx: ToolContext, tool, name: str, input_: dict, tid: str)

def _run_tool(ctx: ToolContext, tool, name: str, input_: dict, tid: str):
    block = ToolUseBlock(id=tid, name=name, input=input_)

    async def drive():
        updates = []
        async for u in run_tool_use(
            block, AssistantMessage(content="t"), _allow, ctx
        ):
            updates.append(u)
        return updates

    return asyncio.run(drive())


def _extract_tool_result(updates: list) -> Any: