A tool whose ``call`` records the abort_controller it received.
(name: str, captured: list, *, concurrent: bool = True)
| 32 | |
| 33 | |
| 34 | def _capture_tool(name: str, captured: list, *, concurrent: bool = True) -> Tool: |
| 35 | """A tool whose ``call`` records the abort_controller it received.""" |
| 36 | |
| 37 | def _call(_inp, ctx): |
| 38 | captured.append(ctx.abort_controller) |
| 39 | return ToolResult(name=name, output=f"ok:{name}") |
| 40 | |
| 41 | return build_tool( |
| 42 | name=name, |
| 43 | input_schema={"type": "object", "properties": {}}, |
| 44 | call=_call, |
| 45 | is_concurrency_safe=lambda _: concurrent, |
| 46 | is_read_only=lambda _: concurrent, |
| 47 | ) |
| 48 | |
| 49 | |
| 50 | def _make_context(tools: list[Tool], abort: AbortController) -> ToolContext: |
no test coverage detected