()
| 279 | |
| 280 | def list_tools(self) -> List[Dict[str, Any]]: |
| 281 | async def fetch(): |
| 282 | if not self._client: |
| 283 | raise RuntimeError("MCP client not connected") |
| 284 | tools = await self._client.list_tools() |
| 285 | # Convert MCP tool objects to OpenAI function format (dicts) |
| 286 | return [ |
| 287 | { |
| 288 | "type": "function", |
| 289 | "function": { |
| 290 | "name": tool.name, |
| 291 | "description": tool.description or "", |
| 292 | "parameters": ( |
| 293 | tool.inputSchema if hasattr(tool, "inputSchema") else {} |
| 294 | ), |
| 295 | }, |
| 296 | } |
| 297 | for tool in tools |
| 298 | ] |
| 299 | |
| 300 | try: |
| 301 | # Use lock to ensure thread-safe access to the event loop |
nothing calls this directly
no test coverage detected