Spaces:
Sleeping
Sleeping
File size: 2,813 Bytes
7d18ad1 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 |
# coding: utf-8
# Copyright (c) 2025 inclusionAI.
import asyncio
from typing import Any, Dict, List, Optional
from mcp.types import CallToolResult, Tool as MCPTool
from aworld.core.common import ActionResult
from aworld.logs.util import logger
from aworld.mcp_client.server import MCPServer
from aworld.tools.function_tools import get_function_tools, FunctionToolsAdapter as BaseAdapter
class FunctionToolsMCPAdapter(MCPServer):
"""Adapter for FunctionTools to MCPServer interface
This adapter allows FunctionTools to be used like a standard MCPServer,
supporting list_tools and call_tool methods.
"""
def __init__(self, name: str):
"""Initialize the adapter
Args:
name: Function tool server name
"""
self._adapter = BaseAdapter(name)
self._name = self._adapter.name
self._connected = False
@property
def name(self) -> str:
"""Server name"""
return self._name
async def connect(self):
"""Connect to the server
For FunctionTools, this is a no-op since no actual connection is needed.
"""
self._connected = True
async def cleanup(self):
"""Clean up server resources
For FunctionTools, this is a no-op since there are no resources to clean up.
"""
self._connected = False
async def list_tools(self) -> List[MCPTool]:
"""List all tools and their descriptions
Returns:
List of tools
"""
if not self._connected:
await self.connect()
# Directly return the tool list from FunctionTools, which now returns MCPTool objects
return await self._adapter.list_tools()
async def call_tool(self, tool_name: str, arguments: Optional[Dict[str, Any]] = None) -> CallToolResult:
"""Call the specified tool function
Args:
tool_name: Tool name
arguments: Tool parameters
Returns:
Tool call result
"""
if not self._connected:
await self.connect()
# Use async method to call the tool
return await self._adapter.call_tool(tool_name, arguments)
def get_function_tools_mcp_adapter(name: str) -> FunctionToolsMCPAdapter:
"""Get MCP adapter for FunctionTools
Args:
name: Function tool server name
Returns:
MCPServer adapter
Raises:
ValueError: When the function tool server with the specified name does not exist
"""
function_tools = get_function_tools(name)
if not function_tools:
raise ValueError(f"FunctionTools '{name}' not found")
return FunctionToolsMCPAdapter(name) |