Upload 3 files

aworld/mcp_client/decorator.py

@@ -0,0 +1,233 @@
+# coding: utf-8
+# Copyright (c) 2025 inclusionAI.
+
+"""
+This module defines decorators for creating MCP servers.
+By using the @mcp_server decorator, you can convert a Python class into an MCP server,
+where the class methods will be automatically converted into MCP tools.
+"""
+
+import inspect
+import functools
+import threading
+from typing import Type, Dict, Any, Optional, Union, List
+
+# Import FastMCP
+from mcp.server import FastMCP
+
+from aworld.core.factory import Factory
+from aworld.logs.util import logger
+
+
+# Save all decorated MCP server classes
+class MCPServerRegistry(Factory):
+    """Register all MCP server classes"""
+
+    def __init__(self, type_name: str = None):
+        super().__init__(type_name)
+        self._instance = {}
+
+    def register(self, name: str, cls: Type, **kwargs):
+        """Register MCP server class"""
+        self._cls[name] = cls
+
+    def get_instance(self, name: str, *args, **kwargs):
+        """Get MCP server instance"""
+        if name not in self._instance:
+            if name not in self._cls:
+                raise ValueError(f"MCP server {name} not registered")
+            self._instance[name] = self._cls[name](*args, **kwargs)
+        return self._instance[name]
+
+
+# Create global registry instance
+MCPServers = MCPServerRegistry()
+
+
+def extract_param_desc(method, param_name):
+    """Extract parameter description from method docstring"""
+    if not method.__doc__:
+        return None
+
+    param_docs = [
+        line.strip() for line in method.__doc__.split('\n')
+        if line.strip().startswith(f":param {param_name}:")
+    ]
+    if param_docs:
+        return param_docs[0].replace(f":param {param_name}:", "").strip()
+    return None
+
+
+def mcp_server(name: str = None, **server_config):
+    """
+    Decorator to convert a class into an MCP server
+
+    Args:
+        name: Server name, if None, uses the class name
+        **server_config: Server configuration parameters
+            - mode: Server running mode, supports 'stdio' and 'sse' (default: 'sse')
+            - host: Host address in SSE mode (default: '127.0.0.1')
+            - port: Port number in SSE mode (default: 8888)
+            - sse_path: Path in SSE mode (default: '/sse')
+            - auto_start: Whether to automatically start the server (default: True)
+
+    Example:
+        @mcp_server(
+            name="simple-calculator",
+            mode="sse",
+            host="localhost",
+            port=8085,
+            sse_path="/calculator/sse"
+        )
+        class Calculator:
+            '''Server description'''
+
+            def __init__(self):
+                self.data = {}
+
+            def get_data(self, key: str) -> str:
+                '''Get data
+                :param key: Data key
+                :return: Data value
+                '''
+                return self.data.get(key, "")
+    """
+    # Extract server configuration or use defaults
+    mode = server_config.get('mode', 'sse')
+    host = server_config.get('host', '127.0.0.1')
+    port = server_config.get('port', 8888)
+    sse_path = server_config.get('sse_path', '/sse')
+    auto_start = server_config.get('auto_start', True)
+
+    def decorator(cls):
+        server_name = name or cls.__name__
+
+        # Use class docstring as server description
+        server_description = cls.__doc__ or f"{server_name} MCP Server"
+
+        # Original initialization method
+        original_init = cls.__init__
+
+        @functools.wraps(original_init)
+        def new_init(self, *args, **kwargs):
+            # Call original initialization method
+            original_init(self, *args, **kwargs)
+
+            # Create FastMCP instance, set server name and description
+            self._mcp = FastMCP(server_name, description=server_description.strip())
+
+            # Tool name list for recording
+            tool_names = []
+
+            # Get all methods, filter out built-in and private methods
+            for method_name, method in inspect.getmembers(self, inspect.ismethod):
+                if not method_name.startswith('_') and method_name != 'run':
+                    # Get method docstring as tool description
+                    tool_description = method.__doc__ or f"{method_name} tool"
+                    tool_description = tool_description.strip()
+
+                    # Record tool name
+                    tool_names.append(method_name)
+
+                    # Create tool and register, using a function generator to ensure each method is correctly bound
+                    def create_tool_wrapper(method_to_call):
+                        # Check if method is async
+                        is_async = inspect.iscoroutinefunction(method_to_call)
+
+                        if is_async:
+                            @self._mcp.tool(name=method_name, description=tool_description)
+                            @functools.wraps(method_to_call)
+                            async def wrapped_method(*args, **kwargs):
+                                return await method_to_call(*args, **kwargs)
+                        else:
+                            @self._mcp.tool(name=method_name, description=tool_description)
+                            @functools.wraps(method_to_call)
+                            def wrapped_method(*args, **kwargs):
+                                return method_to_call(*args, **kwargs)
+
+                        return wrapped_method
+
+                    # Create a dedicated wrapper for each method
+                    create_tool_wrapper(method)
+
+            # Print server information
+            logger.info(f"Creating MCP server: {server_name}")
+            logger.info(f"Server description: {server_description.strip()}")
+            if tool_names:
+                logger.info(f"Registered tools: {', '.join(tool_names)}")
+
+            # Save configuration
+            self._server_config = {
+                'mode': mode,
+                'host': host,
+                'port': port,
+                'sse_path': sse_path
+            }
+
+            # Auto start server if configured
+            if auto_start:
+                # Start server in a new thread to avoid blocking
+                thread = threading.Thread(
+                    target=self.run,
+                    kwargs=self._server_config,
+                    daemon=True
+                )
+                thread.start()
+                logger.info(f"Server {server_name} started in a background thread")
+                self._server_thread = thread
+
+        # Replace initialization method
+        cls.__init__ = new_init
+
+        # Add method to run server
+        def run(self, mode: str = mode, host: str = host, port: int = port, sse_path: str = sse_path):
+            """
+            Run MCP server
+
+            Args:
+                mode: Server running mode, supports 'stdio' and 'sse'
+                host: Host address in SSE mode
+                port: Port number in SSE mode
+                sse_path: Path in SSE mode
+            """
+            if not hasattr(self, '_mcp') or self._mcp is None:
+                raise RuntimeError("MCP server not initialized")
+
+            # Run server according to mode
+            if mode == "stdio":
+                self._mcp.run(transport="stdio")
+            elif mode == "sse":
+                # Configure SSE mode settings
+                self._mcp.settings.host = host
+                self._mcp.settings.port = port
+                self._mcp.settings.sse_path = sse_path
+
+                # Print running information
+                print(f"Running MCP server: {server_name}")
+                print(f"Description: {server_description.strip()}")
+                print(f"Address: http://{host}:{port}{sse_path}")
+
+                self._mcp.run(transport="sse")
+            else:
+                raise ValueError(f"Unsupported mode: {mode}, supported modes are 'stdio' and 'sse'")
+
+        cls.run = run
+
+        # Add a stop method to gracefully stop the server
+        def stop(self):
+            """Stop the MCP server if it's running"""
+            if hasattr(self, '_mcp') and self._mcp is not None:
+                # TODO: Implement proper stopping mechanism based on FastMCP API
+                logger.info(f"Stopping server {server_name}")
+                # Currently there might not be a proper way to stop FastMCP server
+                # This is a placeholder for future implementation
+
+        cls.stop = stop
+
+        # Register to MCP server registry
+        MCPServers.register(server_name, cls)
+
+        # Return modified class
+        return cls
+
+    return decorator 

aworld/mcp_client/server.py

@@ -0,0 +1,330 @@
+from __future__ import annotations
+
+import abc
+import asyncio
+from datetime import timedelta
+import logging
+from contextlib import AbstractAsyncContextManager, AsyncExitStack
+from pathlib import Path
+from typing import Any, Literal
+
+from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
+from mcp import ClientSession, StdioServerParameters, Tool as MCPTool, stdio_client
+from mcp.client.sse import sse_client
+from mcp.types import CallToolResult, JSONRPCMessage
+from typing_extensions import NotRequired, TypedDict
+
+
+class MCPServer(abc.ABC):
+    """Base class for Model Context Protocol servers."""
+
+    @abc.abstractmethod
+    async def connect(self):
+        """Connect to the server. For example, this might mean spawning a subprocess or
+        opening a network connection. The server is expected to remain connected until
+        `cleanup()` is called.
+        """
+        pass
+
+    @property
+    @abc.abstractmethod
+    def name(self) -> str:
+        """A readable name for the server."""
+        pass
+
+    @abc.abstractmethod
+    async def cleanup(self):
+        """Cleanup the server. For example, this might mean closing a subprocess or
+        closing a network connection.
+        """
+        pass
+
+    @abc.abstractmethod
+    async def list_tools(self) -> list[MCPTool]:
+        """List the tools available on the server."""
+        pass
+
+    @abc.abstractmethod
+    async def call_tool(self, tool_name: str, arguments: dict[str, Any] | None) -> CallToolResult:
+        """Invoke a tool on the server."""
+        pass
+
+
+class _MCPServerWithClientSession(MCPServer, abc.ABC):
+    """Base class for MCP servers that use a `ClientSession` to communicate with the server."""
+
+    def __init__(self, cache_tools_list: bool, session_connect_timeout_seconds: int = 30):
+        """
+        Args:
+            cache_tools_list: Whether to cache the tools list. If `True`, the tools list will be
+            cached and only fetched from the server once. If `False`, the tools list will be
+            fetched from the server on each call to `list_tools()`. The cache can be invalidated
+            by calling `invalidate_tools_cache()`. You should set this to `True` if you know the
+            server will not change its tools list, because it can drastically improve latency
+            (by avoiding a round-trip to the server every time).
+
+            session_connect_timeout_seconds: session connect timeout seconds
+        """
+        self.session: ClientSession | None = None
+        self.exit_stack: AsyncExitStack = AsyncExitStack()
+        self._cleanup_lock: asyncio.Lock = asyncio.Lock()
+        self.cache_tools_list = cache_tools_list
+        self.session_connect_timeout_seconds = timedelta(seconds=session_connect_timeout_seconds)
+
+        # The cache is always dirty at startup, so that we fetch tools at least once
+        self._cache_dirty = True
+        self._tools_list: list[MCPTool] | None = None
+
+    @abc.abstractmethod
+    def create_streams(
+            self,
+    ) -> AbstractAsyncContextManager[
+        tuple[
+            MemoryObjectReceiveStream[JSONRPCMessage | Exception],
+            MemoryObjectSendStream[JSONRPCMessage],
+        ]
+    ]:
+        """Create the streams for the server."""
+        pass
+
+    async def __aenter__(self):
+        await self.connect()
+        return self
+
+    async def __aexit__(self, exc_type, exc_value, traceback):
+        await self.cleanup()
+
+    def invalidate_tools_cache(self):
+        """Invalidate the tools cache."""
+        self._cache_dirty = True
+
+    async def connect(self):
+        """Connect to the server."""
+        try:
+            # Ensure closing previous exit_stack to avoid nested async contexts
+            if hasattr(self, 'exit_stack') and self.exit_stack:
+                try:
+                    await self.exit_stack.aclose()
+                except Exception as e:
+                    logging.error(f"Error closing previous exit stack: {e}")
+
+            self.exit_stack = AsyncExitStack()
+
+            # Use a single task context to create the connection
+            transport = await self.exit_stack.enter_async_context(self.create_streams())
+            read, write = transport
+            session = await self.exit_stack.enter_async_context(ClientSession(read, write, read_timeout_seconds=self.session_connect_timeout_seconds))
+            await session.initialize()
+            self.session = session
+        except Exception as e:
+            logging.error(f"Error initializing MCP server: {e}")
+            # Ensure resources are cleaned up if connection fails
+            await self.cleanup()
+            raise
+
+    async def list_tools(self) -> list[MCPTool]:
+        """List the tools available on the server."""
+        if not self.session:
+            raise RuntimeError("Server not initialized. Make sure you call `connect()` first.")
+
+        # Return from cache if caching is enabled, we have tools, and the cache is not dirty
+        if self.cache_tools_list and not self._cache_dirty and self._tools_list:
+            return self._tools_list
+
+        # Reset the cache dirty to False
+        self._cache_dirty = False
+
+        # Fetch the tools from the server
+        self._tools_list = (await self.session.list_tools()).tools
+        return self._tools_list
+
+    async def call_tool(self, tool_name: str, arguments: dict[str, Any] | None) -> CallToolResult:
+        """Invoke a tool on the server."""
+        if not self.session:
+            raise RuntimeError("Server not initialized. Make sure you call `connect()` first.")
+
+        return await self.session.call_tool(tool_name, arguments)
+
+    async def cleanup(self):
+        """Cleanup the server."""
+        async with self._cleanup_lock:
+            try:
+                # Ensure cleanup operations occur in the same task context
+                session = self.session
+                self.session = None  # Remove reference first
+
+                # Wait briefly to ensure any pending operations complete
+                try:
+                    await asyncio.sleep(0.1)
+                except asyncio.CancelledError:
+                    # Ignore cancellation exceptions, continue cleaning resources
+                    pass
+
+                # Clean up exit_stack, ensuring all resources are properly closed
+                exit_stack = self.exit_stack
+                if exit_stack:
+                    try:
+                        await exit_stack.aclose()
+                    except Exception as e:
+                        logging.error(f"Error closing exit stack during cleanup: {e}")
+            except Exception as e:
+                logging.error(f"Error during server cleanup: {e}")
+
+
+class MCPServerStdioParams(TypedDict):
+    """Mirrors `mcp.client.stdio.StdioServerParameters`, but lets you pass params without another
+    import.
+    """
+
+    command: str
+    """The executable to run to start the server. For example, `python` or `node`."""
+
+    args: NotRequired[list[str]]
+    """Command line args to pass to the `command` executable. For example, `['foo.py']` or
+    `['server.js', '--port', '8080']`."""
+
+    env: NotRequired[dict[str, str]]
+    """The environment variables to set for the server. ."""
+
+    cwd: NotRequired[str | Path]
+    """The working directory to use when spawning the process."""
+
+    encoding: NotRequired[str]
+    """The text encoding used when sending/receiving messages to the server. Defaults to `utf-8`."""
+
+    encoding_error_handler: NotRequired[Literal["strict", "ignore", "replace"]]
+    """The text encoding error handler. Defaults to `strict`.
+
+    See https://docs.python.org/3/library/codecs.html#codec-base-classes for
+    explanations of possible values.
+    """
+
+
+class MCPServerStdio(_MCPServerWithClientSession):
+    """MCP server implementation that uses the stdio transport. See the [spec]
+    (https://spec.modelcontextprotocol.io/specification/2024-11-05/basic/transports/#stdio) for
+    details.
+    """
+
+    def __init__(
+            self,
+            params: MCPServerStdioParams,
+            cache_tools_list: bool = False,
+            name: str | None = None,
+    ):
+        """Create a new MCP server based on the stdio transport.
+
+        Args:
+            params: The params that configure the server. This includes the command to run to
+                start the server, the args to pass to the command, the environment variables to
+                set for the server, the working directory to use when spawning the process, and
+                the text encoding used when sending/receiving messages to the server.
+            cache_tools_list: Whether to cache the tools list. If `True`, the tools list will be
+                cached and only fetched from the server once. If `False`, the tools list will be
+                fetched from the server on each call to `list_tools()`. The cache can be
+                invalidated by calling `invalidate_tools_cache()`. You should set this to `True`
+                if you know the server will not change its tools list, because it can drastically
+                improve latency (by avoiding a round-trip to the server every time).
+            name: A readable name for the server. If not provided, we'll create one from the
+                command.
+        """
+        super().__init__(cache_tools_list, int(params.get("env").get("SESSION_REQUEST_CONNECT_TIMEOUT", "60")))
+
+        self.params = StdioServerParameters(
+            command=params["command"],
+            args=params.get("args", []),
+            env=params.get("env"),
+            cwd=params.get("cwd"),
+            encoding=params.get("encoding", "utf-8"),
+            encoding_error_handler=params.get("encoding_error_handler", "strict"),
+        )
+
+        self._name = name or f"stdio: {self.params.command}"
+
+    def create_streams(
+            self,
+    ) -> AbstractAsyncContextManager[
+        tuple[
+            MemoryObjectReceiveStream[JSONRPCMessage | Exception],
+            MemoryObjectSendStream[JSONRPCMessage],
+        ]
+    ]:
+        """Create the streams for the server."""
+        return stdio_client(self.params)
+
+    @property
+    def name(self) -> str:
+        """A readable name for the server."""
+        return self._name
+
+
+class MCPServerSseParams(TypedDict):
+    """Mirrors the params in`mcp.client.sse.sse_client`."""
+
+    url: str
+    """The URL of the server."""
+
+    headers: NotRequired[dict[str, str]]
+    """The headers to send to the server."""
+
+    timeout: NotRequired[float]
+    """The timeout for the HTTP request. Defaults to 60 seconds."""
+
+    sse_read_timeout: NotRequired[float]
+    """The timeout for the SSE connection, in seconds. Defaults to 5 minutes."""
+
+
+class MCPServerSse(_MCPServerWithClientSession):
+    """MCP server implementation that uses the HTTP with SSE transport. See the [spec]
+    (https://spec.modelcontextprotocol.io/specification/2024-11-05/basic/transports/#http-with-sse)
+    for details.
+    """
+
+    def __init__(
+            self,
+            params: MCPServerSseParams,
+            cache_tools_list: bool = False,
+            name: str | None = None,
+    ):
+        """Create a new MCP server based on the HTTP with SSE transport.
+
+        Args:
+            params: The params that configure the server. This includes the URL of the server,
+                the headers to send to the server, the timeout for the HTTP request, and the
+                timeout for the SSE connection.
+
+            cache_tools_list: Whether to cache the tools list. If `True`, the tools list will be
+                cached and only fetched from the server once. If `False`, the tools list will be
+                fetched from the server on each call to `list_tools()`. The cache can be
+                invalidated by calling `invalidate_tools_cache()`. You should set this to `True`
+                if you know the server will not change its tools list, because it can drastically
+                improve latency (by avoiding a round-trip to the server every time).
+
+            name: A readable name for the server. If not provided, we'll create one from the
+                URL.
+        """
+        super().__init__(cache_tools_list)
+
+        self.params = params
+        self._name = name or f"sse: {self.params['url']}"
+
+    def create_streams(
+            self,
+    ) -> AbstractAsyncContextManager[
+        tuple[
+            MemoryObjectReceiveStream[JSONRPCMessage | Exception],
+            MemoryObjectSendStream[JSONRPCMessage],
+        ]
+    ]:
+        """Create the streams for the server."""
+        return sse_client(
+            url=self.params["url"],
+            headers=self.params.get("headers", None),
+            timeout=self.params.get("timeout", 60),
+            sse_read_timeout=self.params.get("sse_read_timeout", 60 * 5),
+        )
+
+    @property
+    def name(self) -> str:
+        """A readable name for the server."""
+        return self._name

aworld/mcp_client/utils.py

@@ -0,0 +1,696 @@
+import logging
+from typing import List, Dict, Any
+import json
+import os
+from contextlib import AsyncExitStack
+import traceback
+
+import requests
+from mcp.types import TextContent, ImageContent
+
+from aworld.core.common import ActionResult
+
+from aworld.logs.util import logger
+from aworld.mcp_client.server import MCPServer, MCPServerSse, MCPServerStdio
+from aworld.tools import get_function_tools
+from aworld.utils.common import find_file
+
+MCP_SERVERS_CONFIG = {}
+
+
+def get_function_tool(sever_name: str) -> List[Dict[str, Any]]:
+    openai_tools = []
+    try:
+        if not sever_name:
+            return []
+        tool_server = get_function_tools(sever_name)
+        if not tool_server:
+            return []
+        tools = tool_server.list_tools()
+        if not tools:
+            return []
+        for tool in tools:
+            required = []
+            properties = {}
+            if tool.inputSchema and tool.inputSchema.get("properties"):
+                required = tool.inputSchema.get("required", [])
+                _properties = tool.inputSchema["properties"]
+                for param_name, param_info in _properties.items():
+                    param_type = (
+                        param_info.get("type")
+                        if param_info.get("type") != "str"
+                        and param_info.get("type") is not None
+                        else "string"
+                    )
+                    param_desc = param_info.get("description", "")
+                    if param_type == "array":
+                        # Handle array type parameters
+                        items_info = param_info.get("items", {})
+                        item_type = items_info.get("type", "string")
+
+                        # Process nested array type parameters
+                        if item_type == "array":
+                            nested_items = items_info.get("items", {})
+                            nested_type = nested_items.get("type", "string")
+
+                            # If the nested type is an object
+                            if nested_type == "object":
+                                properties[param_name] = {
+                                    "description": param_desc,
+                                    "type": param_type,
+                                    "items": {
+                                        "type": item_type,
+                                        "items": {
+                                            "type": nested_type,
+                                            "properties": nested_items.get(
+                                                "properties", {}
+                                            ),
+                                            "required": nested_items.get(
+                                                "required", []
+                                            ),
+                                        },
+                                    },
+                                }
+                            else:
+                                properties[param_name] = {
+                                    "description": param_desc,
+                                    "type": param_type,
+                                    "items": {
+                                        "type": item_type,
+                                        "items": {"type": nested_type},
+                                    },
+                                }
+                        # Process object type cases
+                        elif item_type == "object":
+                            properties[param_name] = {
+                                "description": param_desc,
+                                "type": param_type,
+                                "items": {
+                                    "type": item_type,
+                                    "properties": items_info.get("properties", {}),
+                                    "required": items_info.get("required", []),
+                                },
+                            }
+                        # Process basic type cases
+                        else:
+                            if item_type == "str":
+                                item_type = "string"
+                            properties[param_name] = {
+                                "description": param_desc,
+                                "type": param_type,
+                                "items": {"type": item_type},
+                            }
+                    else:
+                        # Handle non-array type parameters
+                        properties[param_name] = {
+                            "description": param_desc,
+                            "type": param_type,
+                        }
+
+            openai_function_schema = {
+                "name": f"mcp__{sever_name}__{tool.name}",
+                "description": tool.description,
+                "parameters": {
+                    "type": "object",
+                    "properties": properties,
+                    "required": required,
+                },
+            }
+            openai_tools.append(
+                {
+                    "type": "function",
+                    "function": openai_function_schema,
+                }
+            )
+        logging.info(
+            f"✅ function_tool_server #({sever_name}) connected success，tools: {len(tools)}"
+        )
+
+    except Exception as e:
+        logging.warning(
+            f"server_name-get_function_tool:{sever_name} translate failed: {e}"
+        )
+        return []
+    finally:
+        return openai_tools
+
+
+async def run(mcp_servers: list[MCPServer]) -> List[Dict[str, Any]]:
+    openai_tools = []
+    for i, server in enumerate(mcp_servers):
+        try:
+            tools = await server.list_tools()
+            for tool in tools:
+                required = []
+                properties = {}
+                if tool.inputSchema and tool.inputSchema.get("properties"):
+                    required = tool.inputSchema.get("required", [])
+                    _properties = tool.inputSchema["properties"]
+                    for param_name, param_info in _properties.items():
+                        param_type = (
+                            param_info.get("type")
+                            if param_info.get("type") != "str"
+                            and param_info.get("type") is not None
+                            else "string"
+                        )
+                        param_desc = param_info.get("description", "")
+                        if param_type == "array":
+                            # Handle array type parameters
+                            items_info = param_info.get("items", {})
+                            item_type = items_info.get("type", "string")
+
+                            # Process nested array type parameters
+                            if item_type == "array":
+                                nested_items = items_info.get("items", {})
+                                nested_type = nested_items.get("type", "string")
+
+                                # If the nested type is an object
+                                if nested_type == "object":
+                                    properties[param_name] = {
+                                        "description": param_desc,
+                                        "type": param_type,
+                                        "items": {
+                                            "type": item_type,
+                                            "items": {
+                                                "type": nested_type,
+                                                "properties": nested_items.get(
+                                                    "properties", {}
+                                                ),
+                                                "required": nested_items.get(
+                                                    "required", []
+                                                ),
+                                            },
+                                        },
+                                    }
+                                else:
+                                    properties[param_name] = {
+                                        "description": param_desc,
+                                        "type": param_type,
+                                        "items": {
+                                            "type": item_type,
+                                            "items": {"type": nested_type},
+                                        },
+                                    }
+                            # Process object type cases
+                            elif item_type == "object":
+                                properties[param_name] = {
+                                    "description": param_desc,
+                                    "type": param_type,
+                                    "items": {
+                                        "type": item_type,
+                                        "properties": items_info.get("properties", {}),
+                                        "required": items_info.get("required", []),
+                                    },
+                                }
+                            # Process basic type cases
+                            else:
+                                if item_type == "str":
+                                    item_type = "string"
+                                properties[param_name] = {
+                                    "description": param_desc,
+                                    "type": param_type,
+                                    "items": {"type": item_type},
+                                }
+                        else:
+                            # Handle non-array type parameters
+                            properties[param_name] = {
+                                "description": param_desc,
+                                "type": param_type,
+                            }
+
+                openai_function_schema = {
+                    "name": f"{server.name}__{tool.name}",
+                    "description": tool.description,
+                    "parameters": {
+                        "type": "object",
+                        "properties": properties,
+                        "required": required,
+                    },
+                }
+                openai_tools.append(
+                    {
+                        "type": "function",
+                        "function": openai_function_schema,
+                    }
+                )
+            logging.info(
+                f"✅ server #{i + 1} ({server.name}) connected success，tools: {len(tools)}"
+            )
+
+        except Exception as e:
+            logging.error(f"❌ server #{i + 1} ({server.name}) connect fail: {e}")
+            return []
+
+    return openai_tools
+
+
+async def mcp_tool_desc_transform(
+    tools: List[str] = None, mcp_config: Dict[str, Any] = None
+) -> List[Dict[str, Any]]:
+    """Default implement transform framework standard protocol to openai protocol of tool description."""
+    config = {}
+    global MCP_SERVERS_CONFIG
+
+    def _replace_env_variables(config):
+        if isinstance(config, dict):
+            for key, value in config.items():
+                if (
+                    isinstance(value, str)
+                    and value.startswith("${")
+                    and value.endswith("}")
+                ):
+                    env_var_name = value[2:-1]
+                    config[key] = os.getenv(env_var_name, value)
+                    logging.info(f"Replaced {value} with {config[key]}")
+                elif isinstance(value, dict) or isinstance(value, list):
+                    _replace_env_variables(value)
+        elif isinstance(config, list):
+            for index, item in enumerate(config):
+                if (
+                    isinstance(item, str)
+                    and item.startswith("${")
+                    and item.endswith("}")
+                ):
+                    env_var_name = item[2:-1]
+                    config[index] = os.getenv(env_var_name, item)
+                    logging.info(f"Replaced {item} with {config[index]}")
+                elif isinstance(item, dict) or isinstance(item, list):
+                    _replace_env_variables(item)
+
+    if mcp_config:
+        try:
+            config = mcp_config
+            MCP_SERVERS_CONFIG = config
+        except Exception as e:
+            logging.error(f"mcp_config error: {e}")
+            return []
+    else:
+        # Priority given to the running path.
+        config_path = find_file(filename="mcp.json")
+        if not os.path.exists(config_path):
+            current_dir = os.path.dirname(os.path.abspath(__file__))
+            config_path = os.path.normpath(
+                os.path.join(current_dir, "../config/mcp.json")
+            )
+        logger.info(f"mcp conf path: {config_path}")
+
+        if not os.path.exists(config_path):
+            logging.info(f"mcp config is not exist: {config_path}")
+            return []
+
+        try:
+            with open(config_path, "r") as f:
+                config = json.load(f)
+        except Exception as e:
+            logging.info(f"load config fail: {e}")
+            return []
+        _replace_env_variables(config)
+
+        MCP_SERVERS_CONFIG = config
+
+    mcp_servers_config = config.get("mcpServers", {})
+
+    server_configs = []
+    for server_name, server_config in mcp_servers_config.items():
+        # Skip disabled servers
+        if server_config.get("disabled", False):
+            continue
+
+        if tools is None or server_name in tools:
+            # Handle SSE server
+            if "url" in server_config:
+                server_configs.append(
+                    {
+                        "name": "mcp__" + server_name,
+                        "type": "sse",
+                        "params": {"url": server_config["url"]},
+                    }
+                )
+            # Handle stdio server
+            elif "command" in server_config:
+                server_configs.append(
+                    {
+                        "name": "mcp__" + server_name,
+                        "type": "stdio",
+                        "params": {
+                            "command": server_config["command"],
+                            "args": server_config.get("args", []),
+                            "env": server_config.get("env", {}),
+                            "cwd": server_config.get("cwd"),
+                            "encoding": server_config.get("encoding", "utf-8"),
+                            "encoding_error_handler": server_config.get(
+                                "encoding_error_handler", "strict"
+                            ),
+                        },
+                    }
+                )
+
+    if not server_configs:
+        return []
+
+    async with AsyncExitStack() as stack:
+        servers = []
+        for server_config in server_configs:
+            try:
+                if server_config["type"] == "sse":
+                    server = MCPServerSse(
+                        name=server_config["name"], params=server_config["params"]
+                    )
+                elif server_config["type"] == "stdio":
+                    from aworld.mcp_client.server import MCPServerStdio
+
+                    server = MCPServerStdio(
+                        name=server_config["name"], params=server_config["params"]
+                    )
+                else:
+                    logging.warning(
+                        f"Unsupported MCP server type: {server_config['type']}"
+                    )
+                    continue
+
+                server = await stack.enter_async_context(server)
+                servers.append(server)
+            except BaseException as err:
+                # single
+                logging.error(
+                    f"Failed to get tools for MCP server '{server_config['name']}'.\n"
+                    f"Error: {err}\n"
+                    f"Traceback:\n{traceback.format_exc()}"
+                )
+
+        openai_tools = await run(servers)
+
+    return openai_tools
+
+
+async def sandbox_mcp_tool_desc_transform(
+    tools: List[str] = None, mcp_config: Dict[str, Any] = None
+) -> List[Dict[str, Any]]:
+    # todo sandbox mcp_config get from registry
+
+    if not mcp_config:
+        return None
+    config = mcp_config
+    mcp_servers_config = config.get("mcpServers", {})
+    server_configs = []
+    openai_tools = []
+    mcp_openai_tools = []
+
+    for server_name, server_config in mcp_servers_config.items():
+        # Skip disabled servers
+        if server_config.get("disabled", False):
+            continue
+
+        if tools is None or server_name in tools:
+            # Handle SSE server
+            if "function_tool" == server_config.get("type", ""):
+                try:
+                    tmp_function_tool = get_function_tool(server_name)
+                    openai_tools.extend(tmp_function_tool)
+                except Exception as e:
+                    logging.warning(f"server_name:{server_name} translate failed: {e}")
+            elif "api" == server_config.get("type", ""):
+                api_result = requests.get(server_config["url"] + "/list_tools")
+                try:
+                    if not api_result or not api_result.text:
+                        continue
+                        # return None
+                    data = json.loads(api_result.text)
+                    if not data or not data.get("tools"):
+                        continue
+                    for item in data.get("tools"):
+                        tmp_function = {
+                            "type": "function",
+                            "function": {
+                                "name": "mcp__" + server_name + "__" + item["name"],
+                                "description": item["description"],
+                                "parameters": {
+                                    **item["parameters"],
+                                    "properties": {
+                                        k: v
+                                        for k, v in item["parameters"]
+                                        .get("properties", {})
+                                        .items()
+                                        if "default" not in v
+                                    },
+                                },
+                            },
+                        }
+                        openai_tools.append(tmp_function)
+                except Exception as e:
+                    logging.warning(f"server_name:{server_name} translate failed: {e}")
+            elif "sse" == server_config.get("type", ""):
+                server_configs.append(
+                    {
+                        "name": "mcp__" + server_name,
+                        "type": "sse",
+                        "params": {
+                            "url": server_config["url"],
+                            "headers": server_config.get("headers"),
+                        },
+                    }
+                )
+            # Handle stdio server
+            else:
+                # elif "stdio" == server_config.get("type", ""):
+                server_configs.append(
+                    {
+                        "name": "mcp__" + server_name,
+                        "type": "stdio",
+                        "params": {
+                            "command": server_config["command"],
+                            "args": server_config.get("args", []),
+                            "env": server_config.get("env", {}),
+                            "cwd": server_config.get("cwd"),
+                            "encoding": server_config.get("encoding", "utf-8"),
+                            "encoding_error_handler": server_config.get(
+                                "encoding_error_handler", "strict"
+                            ),
+                        },
+                    }
+                )
+
+    if not server_configs:
+        return openai_tools
+
+    async with AsyncExitStack() as stack:
+        servers = []
+        for server_config in server_configs:
+            try:
+                if server_config["type"] == "sse":
+                    server = MCPServerSse(
+                        name=server_config["name"], params=server_config["params"]
+                    )
+                elif server_config["type"] == "stdio":
+                    server = MCPServerStdio(
+                        name=server_config["name"], params=server_config["params"]
+                    )
+                else:
+                    logging.warning(
+                        f"Unsupported MCP server type: {server_config['type']}"
+                    )
+                    continue
+
+                server = await stack.enter_async_context(server)
+                servers.append(server)
+            except BaseException as err:
+                # single
+                logging.error(
+                    f"Failed to get tools for MCP server '{server_config['name']}'.\n"
+                    f"Error: {err}\n"
+                )
+
+        mcp_openai_tools = await run(servers)
+
+    if mcp_openai_tools:
+        openai_tools.extend(mcp_openai_tools)
+
+    return openai_tools
+
+
+async def call_function_tool(
+    server_name: str,
+    tool_name: str,
+    parameter: Dict[str, Any] = None,
+    mcp_config: Dict[str, Any] = None,
+) -> ActionResult:
+    """Specifically handle API type server calls
+
+    Args:
+        server_name: Server name
+        tool_name: Tool name
+        parameter: Parameters
+        mcp_config: MCP configuration
+
+    Returns:
+        ActionResult: Call result
+    """
+    action_result = ActionResult(
+        tool_name=server_name, action_name=tool_name, content="", keep=True
+    )
+    try:
+        tool_server = get_function_tools(server_name)
+        if not tool_server:
+            return action_result
+        call_result_raw = tool_server.call_tool(tool_name, parameter)
+        if call_result_raw and call_result_raw.content:
+            if isinstance(call_result_raw.content[0], TextContent):
+                action_result = ActionResult(
+                    tool_name=server_name,
+                    action_name=tool_name,
+                    content=call_result_raw.content[0].text,
+                    keep=True,
+                    metadata=call_result_raw.content[0].model_extra.get("metadata", {}),
+                )
+            elif isinstance(call_result_raw.content[0], ImageContent):
+                action_result = ActionResult(
+                    tool_name=server_name,
+                    action_name=tool_name,
+                    content=f"data:image/jpeg;base64,{call_result_raw.content[0].data}",
+                    keep=True,
+                    metadata=call_result_raw.content[0].model_extra.get("metadata", {}),
+                )
+
+    except Exception as e:
+        logging.warning(f"call_function_tool ({server_name})({tool_name}) failed: {e}")
+        action_result = ActionResult(
+            tool_name=server_name, action_name=tool_name, content="", keep=True
+        )
+
+    return action_result
+
+
+async def call_api(
+    server_name: str,
+    tool_name: str,
+    parameter: Dict[str, Any] = None,
+    mcp_config: Dict[str, Any] = None,
+) -> ActionResult:
+    """Specifically handle API type server calls
+
+    Args:
+        server_name: Server name
+        tool_name: Tool name
+        parameter: Parameters
+        mcp_config: MCP configuration
+
+    Returns:
+        ActionResult: Call result
+    """
+    action_result = ActionResult(
+        tool_name=server_name, action_name=tool_name, content="", keep=True
+    )
+
+    if not mcp_config or mcp_config.get("mcpServers") is None:
+        return action_result
+
+    mcp_servers = mcp_config.get("mcpServers")
+    if not mcp_servers.get(server_name):
+        return action_result
+
+    server_config = mcp_servers.get(server_name)
+    if "api" != server_config.get("type", ""):
+        logging.warning(
+            f"Server {server_name} is not API type, should use call_tool instead"
+        )
+        return action_result
+
+    try:
+        headers = {"Content-Type": "application/json"}
+        response = requests.post(
+            url=server_config["url"] + "/" + tool_name, headers=headers, json=parameter
+        )
+        action_result = ActionResult(
+            tool_name=server_name,
+            action_name=tool_name,
+            content=response.text,
+            keep=True,
+        )
+    except Exception as e:
+        logging.warning(f"call_api ({server_name})({tool_name}) failed: {e}")
+        action_result = ActionResult(
+            tool_name=server_name,
+            action_name=tool_name,
+            content=f"Error calling API: {str(e)}",
+            keep=True,
+        )
+
+    return action_result
+
+
+async def get_server_instance(
+    server_name: str, mcp_config: Dict[str, Any] = None
+) -> Any:
+    """Get server instance, create a new one if it doesn't exist
+
+    Args:
+        server_name: Server name
+        mcp_config: MCP configuration
+
+    Returns:
+        Server instance or None (if creation fails)
+    """
+    if not mcp_config or mcp_config.get("mcpServers") is None:
+        return None
+
+    mcp_servers = mcp_config.get("mcpServers")
+    if not mcp_servers.get(server_name):
+        return None
+
+    server_config = mcp_servers.get(server_name)
+    try:
+        # API type servers use special handling, no need for persistent connections
+        # Note: We've already handled API type in McpServers.call_tool method
+        # Here we don't return None, but let the caller handle it
+        if "api" == server_config.get("type", ""):
+            logging.info(f"API server {server_name} doesn't need persistent connection")
+            return None
+        elif "sse" == server_config.get("type", ""):
+            server = MCPServerSse(
+                name=server_name,
+                params={
+                    "url": server_config["url"],
+                    "headers": server_config.get("headers"),
+                    "timeout": server_config.get("timeout", 5.0),
+                    "sse_read_timeout": server_config.get("sse_read_timeout", 300.0),
+                },
+            )
+            await server.connect()
+            logging.info(f"Successfully connected to SSE server: {server_name}")
+            return server
+        else:  # stdio type
+            params = {
+                "command": server_config["command"],
+                "args": server_config.get("args", []),
+                "env": server_config.get("env", {}),
+                "cwd": server_config.get("cwd"),
+                "encoding": server_config.get("encoding", "utf-8"),
+                "encoding_error_handler": server_config.get(
+                    "encoding_error_handler", "strict"
+                ),
+            }
+            server = MCPServerStdio(name=server_name, params=params)
+            await server.connect()
+            logging.info(f"Successfully connected to stdio server: {server_name}")
+            return server
+    except Exception as e:
+        logging.warning(f"Failed to create server instance for {server_name}: {e}")
+        return None
+
+
+async def cleanup_server(server):
+    """Clean up server connection
+
+    Args:
+        server: Server instance
+    """
+    try:
+        if hasattr(server, "cleanup"):
+            await server.cleanup()
+        elif hasattr(server, "close"):
+            await server.close()
+        logging.info(
+            f"Successfully cleaned up server: {getattr(server, 'name', 'unknown')}"
+        )
+    except Exception as e:
+        logging.warning(f"Failed to cleanup server: {e}")