Skip to content

Mcp client

Bases: ToolClient

An MCPToolClient connects to an MCP ToolServer and interfaces with its tools

Source code in blue/tools/clients/mcp_client.py 
 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
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
class MCPToolClient(ToolClient):
    """An MCPToolClient connects to an MCP ToolServer and interfaces with its tools"""

    def __init__(self, name, properties={}):
        """Initialize an MCPToolClient instance.

        Parameters:
            name: Name of the tool client
            properties: Properties of the tool client
        """
        super().__init__(name, properties=properties)

    ###### connection
    def _initialize_connection_properties(self):
        """Initialize default connection properties for MCP tool client."""
        super()._initialize_connection_properties()

        # set host, port, protocol
        self.properties['connection']['host'] = 'localhost'
        self.properties['connection']['protocol'] = 'mcp'
        self.properties['connection']['subprotocol'] = 'http'

    ###### connection
    def _connect(self, **connection):
        """Connect to MCP tool server."""
        self._init_connection(**connection)

    def _init_connection(self, **connection):
        """Initialize connection to MCP tool server.
        Parameters:
            connection: Connection parameters including host, port, protocol, subprotocol
        """
        c = copy.deepcopy(connection)
        if 'protocol' in c:
            del c['protocol']

        subprotocol = 'http'
        if 'subprotocol' in c:
            subprotocol = c['subprotocol']
            del c['subprotocol']

        # mcp server url
        host = c['host']
        port = c['port'] if 'port' in c else None

        self.server_url = subprotocol + "://" + host + (":" + str(port) if port else "") + "/mcp"

    async def _create_session(self):
        """Create an MCP client session asynchronously."""
        # Initialize session and client objects
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()

        self._streams_context = streamablehttp_client(
            url=self.server_url,
            headers={},
        )

        read_stream, write_stream, _ = await self._streams_context.__aenter__()

        self._session_context = ClientSession(read_stream, write_stream)
        self.session: ClientSession = await self._session_context.__aenter__()

        await self.session.initialize()

    async def _release_session(self):
        """Release the MCP client session asynchronously."""
        if self._session_context:
            await self._session_context.__aexit__(None, None, None)
        if self._streams_context:  # pylint: disable=W0125
            await self._streams_context.__aexit__(None, None, None)  # pylint: disable=E1101

    def _disconnect(self):
        """Disconnect from MCP tool server."""
        asyncio.run(self._release_session())

    ######### server
    def fetch_metadata(self):
        """Fetch metadata for the MCP tool server.

        Returns:
            An empty dictionary since no metadata is necessary for MCP tool server.
        """
        return {}

    ######### tool
    def fetch_tools(self):
        """Get a list of available tools on MCP tool server.

        Returns:
            List of tool names
        """
        return self.list_tools(detailed=False)

    def fetch_tool_metadata(self, tool):
        """Fetch metadata for a specific tool on MCP tool server.

        Parameters:
            tool: Name of the tool

        Returns:
            Metadata dictionary for the tool
        """
        result = self.list_tools(filter_tools=tool, detailed=True)
        if len(result) == 1:
            return result[0]
        else:
            return {}

    def list_tools(self, filter_tools=None, detailed=True):
        """List available tools on MCP tool server.

        Parameters:
            filter_tools: Tool name or list of tool names to filter. Defaults to None.
            detailed: Whether to return detailed tool information. Defaults to True.

        Returns:
            List of tools optionally with detailed tool information
        """
        return asyncio.run(self._list_tools(filter_tools=filter_tools, detailed=detailed))

    async def _list_tools(self, filter_tools=None, detailed=True):
        """List available tools on MCP tool server asynchronously.

        Parameters:
            filter_tools: Tool name or list of tool names to filter. Defaults to None.
            detailed: Whether to return detailed tool information. Defaults to True.

        Returns:
            List of tools optionally with detailed tool information
        """
        await self._create_session()

        tools = []
        try:
            response = await self.session.list_tools()
            for t in response.tools:
                if detailed:
                    tool = {}
                    tool['name'] = t.name
                    tool['description'] = t.description

                    properties = {}
                    tool['properties'] = properties

                    signature = {}
                    properties['signature'] = signature

                    parameters = {}
                    signature['parameters'] = parameters
                    returns = {'type': 'unknown'}
                    signature['returns'] = returns

                    # process tool schema
                    schema = t.inputSchema
                    required = []
                    if 'required' in t.inputSchema:
                        required = t.inputSchema['required']
                    schema_properties = t.inputSchema['properties']
                    for p in schema_properties:
                        schema_property = schema_properties[p]
                        parameter = {}
                        parameter['type'] = schema_property['type']
                        parameter['required'] = p in required
                        if 'items' in schema_property:
                            parameter['items'] = schema_property['items']
                        parameters[p] = parameter

                    if filter_tools:
                        if type(filter_tools) == str:
                            if t.name == filter_tools:
                                tools.append(tool)
                        elif type(filter_tools) == list:
                            if t.name in filter_tools:
                                tools.append(tool)
                    else:
                        tools.append(tool)
                else:
                    tools.append(t.name)
        finally:
            await self._release_session()
        return tools

    ######### execute tool
    def execute_tool(self, tool, args, kwargs):
        """Execute a specific tool on MCP tool server.

        Parameters:
            tool: Name of the tool
            args: Arguments for the tool function
            kwargs: Keyword arguments for the tool function

        Returns:
            Result of the tool execution
        """
        return asyncio.run(self._execute_tool(tool, args, kwargs))

    async def _execute_tool(self, tool, args, kwargs):
        """Execute a specific tool on MCP tool server asynchronously.

        Parameters:
            tool: Name of the tool
            args: Arguments for the tool function
            kwargs: Keyword arguments for the tool function

        Raises:
            Exception: If no tool is provided

        Returns:
            Result of the tool execution
        """
        if tool is None:
            raise Exception("No tool provided")

        await self._create_session()

        result = []
        try:
            response = await self.session.call_tool(tool, kwargs)
        finally:
            await self._release_session()

        if response:
            contents = response.content
            for content in contents:
                if type(content) == TextContent:
                    result.append(content.text)
        else:
            return []

        return result

__init__(name, properties={})

Initialize an MCPToolClient instance.

Parameters:

Name Type Description Default
name

Name of the tool client

 required
properties

Properties of the tool client

 {}
Source code in blue/tools/clients/mcp_client.py 
28
29
30
31
32
33
34
35
def __init__(self, name, properties={}):
    """Initialize an MCPToolClient instance.

    Parameters:
        name: Name of the tool client
        properties: Properties of the tool client
    """
    super().__init__(name, properties=properties)

execute_tool(tool, args, kwargs)

Execute a specific tool on MCP tool server.

Parameters:

Name Type Description Default
tool

Name of the tool

 required
args

Arguments for the tool function

 required
kwargs

Keyword arguments for the tool function

 required

Returns:

Type Description

Result of the tool execution
Source code in blue/tools/clients/mcp_client.py 
209
210
211
212
213
214
215
216
217
218
219
220
def execute_tool(self, tool, args, kwargs):
    """Execute a specific tool on MCP tool server.

    Parameters:
        tool: Name of the tool
        args: Arguments for the tool function
        kwargs: Keyword arguments for the tool function

    Returns:
        Result of the tool execution
    """
    return asyncio.run(self._execute_tool(tool, args, kwargs))

fetch_metadata()

Fetch metadata for the MCP tool server.

Returns:

Type Description

An empty dictionary since no metadata is necessary for MCP tool server.
Source code in blue/tools/clients/mcp_client.py 
102
103
104
105
106
107
108
def fetch_metadata(self):
    """Fetch metadata for the MCP tool server.

    Returns:
        An empty dictionary since no metadata is necessary for MCP tool server.
    """
    return {}

fetch_tool_metadata(tool)

Fetch metadata for a specific tool on MCP tool server.

Parameters:

Name Type Description Default
tool

Name of the tool

 required

Returns:

Type Description

Metadata dictionary for the tool
Source code in blue/tools/clients/mcp_client.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def fetch_tool_metadata(self, tool):
    """Fetch metadata for a specific tool on MCP tool server.

    Parameters:
        tool: Name of the tool

    Returns:
        Metadata dictionary for the tool
    """
    result = self.list_tools(filter_tools=tool, detailed=True)
    if len(result) == 1:
        return result[0]
    else:
        return {}

fetch_tools()

Get a list of available tools on MCP tool server.

Returns:

Type Description

List of tool names
Source code in blue/tools/clients/mcp_client.py 
111
112
113
114
115
116
117
def fetch_tools(self):
    """Get a list of available tools on MCP tool server.

    Returns:
        List of tool names
    """
    return self.list_tools(detailed=False)

list_tools(filter_tools=None, detailed=True)

List available tools on MCP tool server.

Parameters:

Name Type Description Default
filter_tools

Tool name or list of tool names to filter. Defaults to None.

 None
detailed

Whether to return detailed tool information. Defaults to True.

 True

Returns:

Type Description

List of tools optionally with detailed tool information
Source code in blue/tools/clients/mcp_client.py 
134
135
136
137
138
139
140
141
142
143
144
def list_tools(self, filter_tools=None, detailed=True):
    """List available tools on MCP tool server.

    Parameters:
        filter_tools: Tool name or list of tool names to filter. Defaults to None.
        detailed: Whether to return detailed tool information. Defaults to True.

    Returns:
        List of tools optionally with detailed tool information
    """
    return asyncio.run(self._list_tools(filter_tools=filter_tools, detailed=detailed))
Last update: 2025-10-07