|
| 1 | +""" |
| 2 | +┌──────────────────────────────────────────────────────────────────────────────┐ |
| 3 | +│ @author: Arley Peter │ |
| 4 | +│ @file: mcp_discovery.py │ |
| 5 | +│ Developed by: Arley Peter │ |
| 6 | +│ Creation date: May 05, 2025 │ |
| 7 | +├──────────────────────────────────────────────────────────────────────────────┤ |
| 8 | +│ @copyright © Evolution API 2025. All rights reserved. │ |
| 9 | +│ Licensed under the Apache License, Version 2.0 │ |
| 10 | +│ │ |
| 11 | +│ You may not use this file except in compliance with the License. │ |
| 12 | +│ You may obtain a copy of the License at │ |
| 13 | +│ │ |
| 14 | +│ http://www.apache.org/licenses/LICENSE-2.0 │ |
| 15 | +│ │ |
| 16 | +│ Unless required by applicable law or agreed to in writing, software │ |
| 17 | +│ distributed under the License is distributed on an "AS IS" BASIS, │ |
| 18 | +│ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. │ |
| 19 | +│ See the License for the specific language governing permissions and │ |
| 20 | +│ limitations under the License. │ |
| 21 | +├──────────────────────────────────────────────────────────────────────────────┤ |
| 22 | +│ @important │ |
| 23 | +│ For any future changes to the code in this file, it is recommended to │ |
| 24 | +│ include, together with the modification, the information of the developer │ |
| 25 | +│ who changed it and the date of modification. │ |
| 26 | +└──────────────────────────────────────────────────────────────────────────────┘ |
| 27 | +""" |
| 28 | + |
| 29 | +from typing import List, Dict, Any |
| 30 | +import asyncio |
| 31 | + |
| 32 | +async def _discover_async(config_json: Dict[str, Any]) -> List[Dict[str, Any]]: |
| 33 | + """Return a list[dict] with the tool metadata advertised by the MCP server.""" |
| 34 | + |
| 35 | + from src.services.mcp_service import MCPService |
| 36 | + |
| 37 | + service = MCPService() |
| 38 | + tools, exit_stack = await service._connect_to_mcp_server(config_json) |
| 39 | + serialised = [t.to_dict() if hasattr(t, "to_dict") else { |
| 40 | + "id": t.name, |
| 41 | + "name": t.name, |
| 42 | + "description": getattr(t, "description", t.name), |
| 43 | + "tags": getattr(t, "tags", []), |
| 44 | + "examples": getattr(t, "examples", []), |
| 45 | + "inputModes": getattr(t, "input_modes", ["text"]), |
| 46 | + "outputModes": getattr(t, "output_modes", ["text"]), |
| 47 | + } for t in tools] |
| 48 | + if exit_stack: |
| 49 | + await exit_stack.aclose() |
| 50 | + return serialised |
| 51 | + |
| 52 | + |
| 53 | +def discover_mcp_tools(config_json: Dict[str, Any]) -> List[Dict[str, Any]]: |
| 54 | + """Sync wrapper so we can call it from a sync service function.""" |
| 55 | + return asyncio.run(_discover_async(config_json)) |
0 commit comments