Skip to content

mcp_client_wrapper

MCP client session for Gradio interface.

Provides a client session that connects to MCP servers and processes messages for the Gradio interface, adapted to work with MADA's architecture.

MCPGradioClientSession

MCP client session for Gradio interface.

This class handles connecting to MCP servers and processing messages for the Gradio interface using the existing MADAOrchestrator.

Source code in src/mada/interfaces/gradio/mcp_client_wrapper.py 
 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
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
class MCPGradioClientSession:
    """
    MCP client session for Gradio interface.

    This class handles connecting to MCP servers and processing messages
    for the Gradio interface using the existing MADAOrchestrator.
    """

    def __init__(
        self,
        model_config: ModelConfig,
        agents: List[AgentConfig],
        database_config: DatabaseConfig,
        mcp_servers: MCPServerConfig = None,
    ):
        """
        Initialize the MCP client session.

        Args:
            model_config: Model configuration for MADA
            agents: List of agent configurations
        """
        self.model_config = model_config
        self.agents = agents
        self.database_config = database_config
        self.orchestrator = None
        self.initialized = False
        self.mcp_servers = mcp_servers or {}
        self.session_manager = ChatSessionManager(database_config)
        self.session_bearer_token = None  # Store session bearer token

    async def connect_servers(
        self, agent_table: gr.Dataframe, request: gr.Request
    ) -> Tuple[str, str]:
        """
        Connect to MCP servers based on the agent table configuration.

        Args:
            agent_table: Gradio DataFrame containing agent configurations
            request: Gradio Request object containing HTTP request information.
                Note: This parameter is automatically injected by Gradio when the
                function signature includes `request: gr.Request`. It does not need
                to be added to the additional_inputs list.

        Returns:
            Tuple of (status_message, tools_description)
        """
        # Log request information for debugging/auditing
        if request:
            LOG.info(f"Request headers: {dict(request.headers)}")
            if request.client:
                LOG.info(f"Client IP: {request.client.host}")

            # Store request wormhole community subtoken for use in session
            if hasattr(request, "headers") and "x-subtoken" in request.headers:
                token = request.headers["x-subtoken"]
                self.session_bearer_token = token

        try:
            LOG.info("Starting MCP server connection...")
            if self.orchestrator is None:
                LOG.info("Creating orchestrator instance...")
                self.orchestrator = MADAOrchestrator(
                    self.model_config,
                    self.database_config,
                    session_manager=self.session_manager,
                    bearer_token=self.session_bearer_token,
                )
                # Enter the async context manager (required for proper setup)
                LOG.info("Entering orchestrator async context...")
                await self.orchestrator.__aenter__()
                LOG.info("Orchestrator context entered successfully")

            # Initialize orchestrator with agents
            LOG.info(
                f"Initializing orchestrator with {len(self.agents)} agents and {len(self.mcp_servers)} MCP servers..."
            )
            status_msg, tools = await self.orchestrator.initialize_orchestrator(
                agent_configs=self.agents,  # Use provided agents
                mcp_servers=self.mcp_servers,  # Placeholder for MCP server config, replace with real config when available
            )
            LOG.info("Orchestrator initialization complete!")

            agent_dict = cycle_through_tools(self.orchestrator.specialist_agents)
            self.initialized = True
            return gr.Button(status_msg, elem_id="green_btn"), create_agent_table(
                self.agents, agent_dict
            )

        except BaseExceptionGroup as eg:
            error_msg = f"Failed to connect to MCP servers: {len(eg.exceptions)} error(s) occurred"
            LOG.error(error_msg, exc_info=True)
            return error_msg, f"Connection failed: {len(eg.exceptions)} errors"
        except Exception as e:
            error_msg = f"Failed to connect to MCP servers: {e}"
            LOG.error(error_msg)
            LOG.error("Full traceback:", exc_info=True)
            return gr.Button(error_msg, variant="stop"), create_agent_table(self.agents)

    def list_sessions(self) -> List[str]:
        """
        List all sessions available and create a list of labels for each session.

        Returns:
            A list of labels of the form "timestamp | ID" for each session.
        """
        sessions = self.session_manager.list_sessions()
        return [
            f"{ts.isoformat(sep=' ', timespec='seconds')} | {sid}"
            for sid, ts in sessions
        ]

    def get_session_choices(self) -> List[str]:
        """
        List all chat sessions.

        This is used on startup of the Gradio app.

        Returns:
            A list of chat session labels of the form "timestamp | session ID".
        """
        return self.list_sessions()

    def update_session_choices(self) -> gr.update:
        """
        Update the list of chat sessions.

        Returns:
            A gradio update object with new chat session choices.
        """
        return gr.update(choices=self.list_sessions(), value=None)

    def create_new_session(self) -> Tuple[gr.update, List]:
        """
        Create a new chat session.

        Returns:
            A tuple containing a gradio update object for updating the
                sessions list and an empty chat history.
        """
        new_id = self.session_manager.create_session_id()
        self.session_manager.create_new_session(new_id)
        self.session_manager.select_session(new_id)
        updated_sessions = self.list_sessions()
        return gr.update(
            choices=updated_sessions, value=None
        ), []  # update sessions list, empty chat history

    def _extract_id_from_label(self, session_label: str) -> str:
        """
        Extract session id from label.

        Args:
            session_label (str): The label of the session of the form "timestamp | ID"

        Returns:
            The session ID.
        """
        try:
            session_id = session_label.split("|", 1)[1].strip()
        except Exception:
            session_id = session_label.strip()

        return session_id

    def select_session(self, session_label: str) -> List[Dict[str, str]]:
        """
        Given a label like 'timestamp | session_id', select that session and return its history.

        Args:
            session_label (str): The label of the chat session that the user wants to select.

        Returns:
            A list of the chat history.
        """
        if not session_label:
            return []  # nothing selected

        # Extract session id from label
        session_id = self._extract_id_from_label(session_label)

        history = self.session_manager.select_session(session_id)

        return history

    def delete_session(self, session_label: str) -> Tuple[gr.update, List]:
        """
        Delete a chat session.

        Args:
            session_label (str): The label of the chat session that the user wants to delete.

        Returns:
            A tuple containing a gradio update object for updating the
                sessions list and an empty chat history.
        """
        # Handle case where no session is selected
        if not session_label:
            # Return current state unchanged
            updated_sessions = self.list_sessions()
            return gr.update(choices=updated_sessions, value=None), []

        session_id = self._extract_id_from_label(session_label)
        self.session_manager.delete_session(session_id)
        updated_sessions = self.list_sessions()
        return gr.update(
            choices=updated_sessions, value=None
        ), []  # update sessions list, clear chat history

    def delete_all_sessions(self) -> Tuple[gr.update, List]:
        """
        Delete all chat sessions with proper error handling.

        Returns:
            A tuple containing:
            - gradio update object for updating the sessions list
            - chat history (empty if successful, unchanged if error)
        """
        try:
            LOG.info("Attempting to delete all sessions")
            self.session_manager.delete_all_sessions(confirm=False)
            LOG.info("Successfully deleted all sessions")
            return gr.update(choices=[], value=None), []
        except Exception as e:
            LOG.error(f"Failed to delete all sessions: {e}")
            # Return current state unchanged if deletion fails
            current_sessions = self.list_sessions()
            return gr.update(choices=current_sessions), []

    async def process_message(
        self,
        message: str,
        history: List,
        agent_table: gr.Dataframe,
    ) -> AsyncGenerator[str, None]:
        """
        Process a user message through the orchestrator.

        Args:
            message: User input message
            history: Chat history (not used in current implementation)
            agent_table: Agent configuration table (not used in current implementation)

        Yields:
            Response chunks from the orchestrator
        """
        if not self.initialized:
            yield "Error: MCP servers not connected. Please connect first."
            return

        if not self.orchestrator:
            yield "Error: Orchestrator not initialized."
            return

        try:
            # Process message through orchestrator
            full_response = ""
            async for chunk in self.orchestrator.process_message(message):
                full_response += chunk

            yield full_response

        except Exception as e:
            error_msg = f"Error processing message: {e}"
            LOG.error(error_msg)
            traceback.print_exc()
            yield error_msg

    async def cleanup(self):
        """Clean up resources."""
        if self.orchestrator:
            try:
                await self.orchestrator.__aexit__(None, None, None)
            except BaseExceptionGroup as eg:
                LOG.warning(
                    f"Multiple errors during cleanup ({len(eg.exceptions)} errors, suppressed)"
                )
            except Exception as e:
                LOG.error(f"Error during cleanup: {e}")
        self.initialized = False

__init__(model_config, agents, database_config, mcp_servers=None)

Initialize the MCP client session.

Parameters:

Name Type Description Default
model_config ModelConfig

Model configuration for MADA

 required
agents List[AgentConfig]

List of agent configurations

 required
Source code in src/mada/interfaces/gradio/mcp_client_wrapper.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def __init__(
    self,
    model_config: ModelConfig,
    agents: List[AgentConfig],
    database_config: DatabaseConfig,
    mcp_servers: MCPServerConfig = None,
):
    """
    Initialize the MCP client session.

    Args:
        model_config: Model configuration for MADA
        agents: List of agent configurations
    """
    self.model_config = model_config
    self.agents = agents
    self.database_config = database_config
    self.orchestrator = None
    self.initialized = False
    self.mcp_servers = mcp_servers or {}
    self.session_manager = ChatSessionManager(database_config)
    self.session_bearer_token = None  # Store session bearer token

cleanup() async

Clean up resources.

Source code in src/mada/interfaces/gradio/mcp_client_wrapper.py 
299
300
301
302
303
304
305
306
307
308
309
310
async def cleanup(self):
    """Clean up resources."""
    if self.orchestrator:
        try:
            await self.orchestrator.__aexit__(None, None, None)
        except BaseExceptionGroup as eg:
            LOG.warning(
                f"Multiple errors during cleanup ({len(eg.exceptions)} errors, suppressed)"
            )
        except Exception as e:
            LOG.error(f"Error during cleanup: {e}")
    self.initialized = False

connect_servers(agent_table, request) async

Connect to MCP servers based on the agent table configuration.

Parameters:

Name Type Description Default
agent_table Dataframe

Gradio DataFrame containing agent configurations

 required
request Request

Gradio Request object containing HTTP request information. Note: This parameter is automatically injected by Gradio when the function signature includes request: gr.Request. It does not need to be added to the additional_inputs list.

 required

Returns:

Type Description
Tuple[str, str]

Tuple of (status_message, tools_description)
Source code in src/mada/interfaces/gradio/mcp_client_wrapper.py 
 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
async def connect_servers(
    self, agent_table: gr.Dataframe, request: gr.Request
) -> Tuple[str, str]:
    """
    Connect to MCP servers based on the agent table configuration.

    Args:
        agent_table: Gradio DataFrame containing agent configurations
        request: Gradio Request object containing HTTP request information.
            Note: This parameter is automatically injected by Gradio when the
            function signature includes `request: gr.Request`. It does not need
            to be added to the additional_inputs list.

    Returns:
        Tuple of (status_message, tools_description)
    """
    # Log request information for debugging/auditing
    if request:
        LOG.info(f"Request headers: {dict(request.headers)}")
        if request.client:
            LOG.info(f"Client IP: {request.client.host}")

        # Store request wormhole community subtoken for use in session
        if hasattr(request, "headers") and "x-subtoken" in request.headers:
            token = request.headers["x-subtoken"]
            self.session_bearer_token = token

    try:
        LOG.info("Starting MCP server connection...")
        if self.orchestrator is None:
            LOG.info("Creating orchestrator instance...")
            self.orchestrator = MADAOrchestrator(
                self.model_config,
                self.database_config,
                session_manager=self.session_manager,
                bearer_token=self.session_bearer_token,
            )
            # Enter the async context manager (required for proper setup)
            LOG.info("Entering orchestrator async context...")
            await self.orchestrator.__aenter__()
            LOG.info("Orchestrator context entered successfully")

        # Initialize orchestrator with agents
        LOG.info(
            f"Initializing orchestrator with {len(self.agents)} agents and {len(self.mcp_servers)} MCP servers..."
        )
        status_msg, tools = await self.orchestrator.initialize_orchestrator(
            agent_configs=self.agents,  # Use provided agents
            mcp_servers=self.mcp_servers,  # Placeholder for MCP server config, replace with real config when available
        )
        LOG.info("Orchestrator initialization complete!")

        agent_dict = cycle_through_tools(self.orchestrator.specialist_agents)
        self.initialized = True
        return gr.Button(status_msg, elem_id="green_btn"), create_agent_table(
            self.agents, agent_dict
        )

    except BaseExceptionGroup as eg:
        error_msg = f"Failed to connect to MCP servers: {len(eg.exceptions)} error(s) occurred"
        LOG.error(error_msg, exc_info=True)
        return error_msg, f"Connection failed: {len(eg.exceptions)} errors"
    except Exception as e:
        error_msg = f"Failed to connect to MCP servers: {e}"
        LOG.error(error_msg)
        LOG.error("Full traceback:", exc_info=True)
        return gr.Button(error_msg, variant="stop"), create_agent_table(self.agents)

create_new_session()

Create a new chat session.

Returns:

Type Description
Tuple[update, List]

A tuple containing a gradio update object for updating the sessions list and an empty chat history.
Source code in src/mada/interfaces/gradio/mcp_client_wrapper.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
def create_new_session(self) -> Tuple[gr.update, List]:
    """
    Create a new chat session.

    Returns:
        A tuple containing a gradio update object for updating the
            sessions list and an empty chat history.
    """
    new_id = self.session_manager.create_session_id()
    self.session_manager.create_new_session(new_id)
    self.session_manager.select_session(new_id)
    updated_sessions = self.list_sessions()
    return gr.update(
        choices=updated_sessions, value=None
    ), []  # update sessions list, empty chat history

delete_all_sessions()

Delete all chat sessions with proper error handling.

Returns:

Type Description
update

A tuple containing:
List
  • gradio update object for updating the sessions list
Tuple[update, List]
  • chat history (empty if successful, unchanged if error)
Source code in src/mada/interfaces/gradio/mcp_client_wrapper.py 
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
def delete_all_sessions(self) -> Tuple[gr.update, List]:
    """
    Delete all chat sessions with proper error handling.

    Returns:
        A tuple containing:
        - gradio update object for updating the sessions list
        - chat history (empty if successful, unchanged if error)
    """
    try:
        LOG.info("Attempting to delete all sessions")
        self.session_manager.delete_all_sessions(confirm=False)
        LOG.info("Successfully deleted all sessions")
        return gr.update(choices=[], value=None), []
    except Exception as e:
        LOG.error(f"Failed to delete all sessions: {e}")
        # Return current state unchanged if deletion fails
        current_sessions = self.list_sessions()
        return gr.update(choices=current_sessions), []

delete_session(session_label)

Delete a chat session.

Parameters:

Name Type Description Default
session_label str

The label of the chat session that the user wants to delete.

 required

Returns:

Type Description
Tuple[update, List]

A tuple containing a gradio update object for updating the sessions list and an empty chat history.
Source code in src/mada/interfaces/gradio/mcp_client_wrapper.py 
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
def delete_session(self, session_label: str) -> Tuple[gr.update, List]:
    """
    Delete a chat session.

    Args:
        session_label (str): The label of the chat session that the user wants to delete.

    Returns:
        A tuple containing a gradio update object for updating the
            sessions list and an empty chat history.
    """
    # Handle case where no session is selected
    if not session_label:
        # Return current state unchanged
        updated_sessions = self.list_sessions()
        return gr.update(choices=updated_sessions, value=None), []

    session_id = self._extract_id_from_label(session_label)
    self.session_manager.delete_session(session_id)
    updated_sessions = self.list_sessions()
    return gr.update(
        choices=updated_sessions, value=None
    ), []  # update sessions list, clear chat history

get_session_choices()

List all chat sessions.

This is used on startup of the Gradio app.

Returns:

Type Description
List[str]

A list of chat session labels of the form "timestamp | session ID".
Source code in src/mada/interfaces/gradio/mcp_client_wrapper.py 
143
144
145
146
147
148
149
150
151
152
def get_session_choices(self) -> List[str]:
    """
    List all chat sessions.

    This is used on startup of the Gradio app.

    Returns:
        A list of chat session labels of the form "timestamp | session ID".
    """
    return self.list_sessions()

list_sessions()

List all sessions available and create a list of labels for each session.

Returns:

Type Description
List[str]

A list of labels of the form "timestamp | ID" for each session.
Source code in src/mada/interfaces/gradio/mcp_client_wrapper.py 
130
131
132
133
134
135
136
137
138
139
140
141
def list_sessions(self) -> List[str]:
    """
    List all sessions available and create a list of labels for each session.

    Returns:
        A list of labels of the form "timestamp | ID" for each session.
    """
    sessions = self.session_manager.list_sessions()
    return [
        f"{ts.isoformat(sep=' ', timespec='seconds')} | {sid}"
        for sid, ts in sessions
    ]

process_message(message, history, agent_table) async

Process a user message through the orchestrator.

Parameters:

Name Type Description Default
message str

User input message

 required
history List

Chat history (not used in current implementation)

 required
agent_table Dataframe

Agent configuration table (not used in current implementation)

 required

Yields:

Type Description
AsyncGenerator[str, None]

Response chunks from the orchestrator
Source code in src/mada/interfaces/gradio/mcp_client_wrapper.py 
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
async def process_message(
    self,
    message: str,
    history: List,
    agent_table: gr.Dataframe,
) -> AsyncGenerator[str, None]:
    """
    Process a user message through the orchestrator.

    Args:
        message: User input message
        history: Chat history (not used in current implementation)
        agent_table: Agent configuration table (not used in current implementation)

    Yields:
        Response chunks from the orchestrator
    """
    if not self.initialized:
        yield "Error: MCP servers not connected. Please connect first."
        return

    if not self.orchestrator:
        yield "Error: Orchestrator not initialized."
        return

    try:
        # Process message through orchestrator
        full_response = ""
        async for chunk in self.orchestrator.process_message(message):
            full_response += chunk

        yield full_response

    except Exception as e:
        error_msg = f"Error processing message: {e}"
        LOG.error(error_msg)
        traceback.print_exc()
        yield error_msg

select_session(session_label)

Given a label like 'timestamp | session_id', select that session and return its history.

Parameters:

Name Type Description Default
session_label str

The label of the chat session that the user wants to select.

 required

Returns:

Type Description
List[Dict[str, str]]

A list of the chat history.
Source code in src/mada/interfaces/gradio/mcp_client_wrapper.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
def select_session(self, session_label: str) -> List[Dict[str, str]]:
    """
    Given a label like 'timestamp | session_id', select that session and return its history.

    Args:
        session_label (str): The label of the chat session that the user wants to select.

    Returns:
        A list of the chat history.
    """
    if not session_label:
        return []  # nothing selected

    # Extract session id from label
    session_id = self._extract_id_from_label(session_label)

    history = self.session_manager.select_session(session_id)

    return history

update_session_choices()

Update the list of chat sessions.

Returns:

Type Description
update

A gradio update object with new chat session choices.
Source code in src/mada/interfaces/gradio/mcp_client_wrapper.py 
154
155
156
157
158
159
160
161
def update_session_choices(self) -> gr.update:
    """
    Update the list of chat sessions.

    Returns:
        A gradio update object with new chat session choices.
    """
    return gr.update(choices=self.list_sessions(), value=None)