Handle tool execution.
(name: str, arguments: dict | None)
| 31 | |
| 32 | @mcp_server.call_tool() |
| 33 | async def handle_call_tool(name: str, arguments: dict | None) -> list[TextContent]: |
| 34 | """Handle tool execution.""" |
| 35 | server = get_server() |
| 36 | args = arguments or {} |
| 37 | |
| 38 | result = await server.handle_tool_call(name, args) |
| 39 | result = _strip_workspace_prefix(result) |
| 40 | |
| 41 | if "error" in result: |
| 42 | return [TextContent(type="text", text=f"Error: {result['error']}")] |
| 43 | |
| 44 | response_text = json.dumps(result, indent=2) |
| 45 | response_text = _apply_response_token_limit(name, response_text) |
| 46 | return [TextContent(type="text", text=response_text)] |
| 47 | |
| 48 | # Create the SSE transport. |
| 49 | sse = SseServerTransport("/api/v1/mcp/messages") |
nothing calls this directly
no test coverage detected