Implementation:EvolvingLMMs Lab Lmms eval MCP Client

Knowledge Sources	EvolvingLMMs_Lab_Lmms_eval
Domains	Tool Integration, MCP Protocol
Last Updated	2026-02-14 00:00 GMT

Overview

The MCP Client provides a Python client for connecting to and invoking Model Context Protocol (MCP) servers. It handles server lifecycle management, tool discovery, tool execution, and format conversion between MCP content types and OpenAI-compatible message formats.

Description

The MCPClient class communicates with MCP servers over stdio, launching a server subprocess for each operation. It offers async methods for discovering available tools (get_function_list) and executing them (run_tool), along with synchronous wrappers that manage their own event loops. A convert_result_to_openai_format method translates MCP ImageContent, TextContent, and AudioContent into the OpenAI message content format used by the evaluation framework.

Usage

Use the MCP Client when a task or model needs to invoke external tools exposed via MCP servers (e.g., video cropping, image processing). The client discovers tool schemas for inclusion in model prompts and executes tool calls returned by the model.

Code Reference

Source Location

• Repository: EvolvingLMMs-Lab/lmms-eval
• File: lmms_eval/mcp/client.py
• Lines: 1--96

Key Components

MCPClient.__init__

class MCPClient:
    def __init__(self, server_path: str, timeout: timedelta = timedelta(seconds=600)):
        self.server_path = server_path
        self.timeout = timeout

Purpose: Initialize the client with a path to the MCP server script and an optional timeout.

Parameters:

• server_path -- Path to the MCP server Python script
• timeout -- Timeout for server operations (default: 10 minutes)

get_function_list

async def get_function_list(self):
    server_params = StdioServerParameters(command="python", args=[self.server_path])
    async with stdio_client(server=server_params) as (read_stream, write_stream):
        async with ClientSession(
            read_stream, write_stream, read_timeout_seconds=self.timeout
        ) as session:
            await session.initialize()
            tools = (await session.list_tools()).tools
            functions = []
            for tool in tools:
                functions.append({
                    "type": "function",
                    "function": {
                        "name": tool.name,
                        "description": tool.description or "",
                        "parameters": tool.inputSchema,
                    },
                })
            return functions

Purpose: Discover available tools from the MCP server and return them in OpenAI function-calling format.

Returns: List of tool schemas in OpenAI-compatible format, each containing name, description, and parameters.

run_tool

async def run_tool(self, tool_name: str, tool_args: dict):
    server_params = StdioServerParameters(command="python", args=[self.server_path])
    async with stdio_client(server=server_params) as (read_stream, write_stream):
        async with ClientSession(
            read_stream, write_stream, read_timeout_seconds=self.timeout
        ) as session:
            await session.initialize()
            result = await session.call_tool(tool_name, tool_args)
            return result

Purpose: Execute a specific tool on the MCP server.

Parameters:

• tool_name -- Name of the tool to invoke
• tool_args -- Dictionary of arguments to pass to the tool

Returns: Tool execution result containing MCP content objects.

convert_result_to_openai_format

def convert_result_to_openai_format(
    self,
    result: Union[
        ImageContent, TextContent, AudioContent,
        List[Union[ImageContent, TextContent, AudioContent]],
    ],
) -> dict:
    if isinstance(result, list):
        results = []
        for item in result:
            results.append(self.convert_result_to_openai_format(item))
        return results
    if isinstance(result, ImageContent):
        return [{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{result.data}"}}]
    elif isinstance(result, TextContent):
        return [{"type": "text", "text": result.text}]
    elif isinstance(result, AudioContent):
        return [{"type": "audio_url", "audio_url": {"url": f"data:audio/wav;base64,{result.data}"}}]
    else:
        raise ValueError(f"Unsupported result type : {type(result)}")

Purpose: Convert MCP content types to OpenAI message format.

Supported Conversions:

• ImageContent -- Converted to image_url with base64 data URI
• TextContent -- Converted to text content block
• AudioContent -- Converted to audio_url with base64 data URI

Returns: List of OpenAI-compatible content blocks, or a list of lists when the input is a list.

get_function_list_sync

def get_function_list_sync(self):
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        return loop.run_until_complete(self.get_function_list())
    finally:
        loop.close()

Purpose: Synchronous wrapper for get_function_list. Creates a new event loop, runs the async method to completion, and cleans up.

run_tool_sync

def run_tool_sync(self, tool_name: str, tool_args: dict):
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        return loop.run_until_complete(self.run_tool(tool_name, tool_args))
    finally:
        loop.close()

Purpose: Synchronous wrapper for run_tool. Creates a new event loop, runs the async method to completion, and cleans up.

Parameters:

• tool_name -- Name of the tool to invoke
• tool_args -- Dictionary of arguments to pass to the tool

I/O Contract

Input	Type	Description
server_path	str	Path to MCP server Python script
timeout	timedelta	Maximum time for server operations
tool_name	str	Name of the tool to execute
tool_args	dict	Arguments to pass to the tool

Output	Type	Description
functions	list[dict]	Tool schemas in OpenAI function-calling format
result	MCP result	Raw tool execution result with content objects
openai_content	list[dict]	Content blocks in OpenAI message format

Integration Example

from lmms_eval.mcp.client import MCPClient
from datetime import timedelta

# Initialize client
client = MCPClient("/path/to/mcp_server.py", timeout=timedelta(seconds=300))

# Discover available tools
functions = client.get_function_list_sync()

# Include tool schemas in model request
response = model.generate(
    messages=[{"role": "user", "content": "Process this video"}],
    tools=functions,
)

# Execute tool calls returned by the model
if response.tool_calls:
    for call in response.tool_calls:
        result = client.run_tool_sync(call.function.name, call.function.arguments)
        openai_content = client.convert_result_to_openai_format(result.content)

Dependencies

• asyncio -- Async/await support and event loops
• datetime.timedelta -- Timeout specification
• typing -- Type annotations
• mcp.ClientSession -- Session management
• mcp.StdioServerParameters -- Server configuration
• mcp.client.stdio.stdio_client -- stdio transport context manager
• mcp.types -- AudioContent, ImageContent, TextContent

Design Decisions

• Stdio communication -- Uses stdin/stdout for server communication; simple, language-agnostic, and requires no network configuration.
• Async primary API -- Core methods are async for non-blocking I/O; synchronous wrappers provided for convenience.
• Short-lived sessions -- A new server subprocess is launched per operation, simplifying lifecycle management and avoiding state leakage between calls at the cost of higher overhead.
• OpenAI format conversion -- Translates MCP content types to the standard OpenAI message format for interoperability with model APIs.
• Event loop per sync call -- Creates a new event loop each time to avoid conflicts with any existing loop, ensuring clean state.
• Configurable timeout -- Per-client timeout accommodates tools with varying runtime characteristics (default 10 minutes).

Limitations

• A new server process is spawned per call, which is inefficient for many rapid calls.
• No connection pooling or persistent server connections.
• Server state is not preserved between calls.
• No streaming support; results are returned only after completion.
• The server command is hardcoded to "python".

Related Pages

• Principle: EvolvingLMMs_Lab_Lmms_eval_MCP_Tool_Integration
• See Also: EvolvingLMMs_Lab_Lmms_eval_Crop_Video_MCP_Server, EvolvingLMMs_Lab_Lmms_eval_Sample_MCP_Server