MCP Web Tools Server
by surya-madhav
- frontend
- pages
import os
import json
import streamlit as st
import asyncio
import sys
# Add the parent directory to the Python path to import utils
sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
from frontend.utils import load_config, connect_to_server, call_tool, default_config_path
st.title("My Active MCP Servers")
# Configuration and server selection in the sidebar
with st.sidebar:
st.subheader("Configuration")
# Config file path input with unique key
config_path = st.text_input(
"Path to config file",
value=st.session_state.get('config_path', default_config_path),
key="config_path_input_sidebar"
)
# Update the session state with the new path
st.session_state.config_path = config_path
if st.button("Load Servers", key="load_servers_sidebar"):
if os.path.exists(config_path):
config_data = load_config(config_path)
if config_data and 'mcpServers' in config_data:
st.session_state.config_data = config_data
st.session_state.servers = config_data.get('mcpServers', {})
# Add debug message
message = f"Found {len(st.session_state.servers)} MCP servers in config"
if 'debug_messages' in st.session_state:
st.session_state.debug_messages.append(message)
st.success(message)
else:
error_msg = "No MCP servers found in config"
if 'debug_messages' in st.session_state:
st.session_state.debug_messages.append(error_msg)
st.error(error_msg)
else:
error_msg = f"Config file not found: {config_path}"
if 'debug_messages' in st.session_state:
st.session_state.debug_messages.append(error_msg)
st.error(error_msg)
# Server selection dropdown
st.divider()
st.subheader("Server Selection")
if 'servers' in st.session_state and st.session_state.servers:
server_names = list(st.session_state.servers.keys())
selected_server = st.selectbox(
"Select an MCP server",
server_names,
key="server_selection_sidebar"
)
if st.button("Connect", key="connect_button_sidebar"):
server_config = st.session_state.servers.get(selected_server, {})
command = server_config.get('command')
args = server_config.get('args', [])
env = server_config.get('env', {})
with st.spinner(f"Connecting to {selected_server}..."):
# Add debug message
debug_msg = f"Connecting to {selected_server}..."
if 'debug_messages' in st.session_state:
st.session_state.debug_messages.append(debug_msg)
# Connect to the server
server_info = asyncio.run(connect_to_server(command, args, env))
st.session_state[f'server_info_{selected_server}'] = server_info
st.session_state.active_server = selected_server
# Add debug message about connection success/failure
if server_info.get('tools'):
success_msg = f"Connected to {selected_server}: {len(server_info['tools'])} tools"
if 'debug_messages' in st.session_state:
st.session_state.debug_messages.append(success_msg)
else:
error_msg = f"Connected but no tools found"
if 'debug_messages' in st.session_state:
st.session_state.debug_messages.append(error_msg)
# Force the page to refresh to show connected server details
st.rerun()
else:
st.info("Load config to see servers")
# Main area: Only display content when a server is connected
if 'active_server' in st.session_state and st.session_state.active_server:
active_server = st.session_state.active_server
server_info_key = f'server_info_{active_server}'
if server_info_key in st.session_state:
st.subheader(f"Connected to: {active_server}")
server_info = st.session_state[server_info_key]
server_config = st.session_state.servers.get(active_server, {})
# Display server configuration
with st.expander("Server Configuration"):
st.json(server_config)
# Display tools
if server_info.get('tools'):
st.subheader("Available Tools")
# Create tabs for each tool
tool_tabs = st.tabs([tool.name for tool in server_info['tools']])
for i, tool in enumerate(server_info['tools']):
with tool_tabs[i]:
st.markdown(f"**Description:** {tool.description or 'No description provided'}")
# Tool schema
if hasattr(tool, 'inputSchema') and tool.inputSchema:
with st.expander("Input Schema"):
st.json(tool.inputSchema)
# Generate form for tool inputs
st.subheader("Call Tool")
# Create a form
with st.form(key=f"tool_form_{active_server}_{tool.name}"):
# Fix duplicate ID error by adding unique keys for form fields
tool_inputs = {}
# Check if input schema has properties
if 'properties' in tool.inputSchema:
# Create form inputs based on schema properties
for param_name, param_schema in tool.inputSchema['properties'].items():
param_type = param_schema.get('type', 'string')
# Create unique key for each form field
field_key = f"{active_server}_{tool.name}_{param_name}"
if param_type == 'string':
tool_inputs[param_name] = st.text_input(
f"{param_name}",
help=param_schema.get('description', ''),
key=field_key
)
elif param_type == 'number' or param_type == 'integer':
tool_inputs[param_name] = st.number_input(
f"{param_name}",
help=param_schema.get('description', ''),
key=field_key
)
elif param_type == 'boolean':
tool_inputs[param_name] = st.checkbox(
f"{param_name}",
help=param_schema.get('description', ''),
key=field_key
)
# Add more types as needed
# Submit button
submit_button = st.form_submit_button(f"Execute {tool.name}")
if submit_button:
# Get server config
command = server_config.get('command')
args = server_config.get('args', [])
with st.spinner(f"Executing {tool.name}..."):
# Add debug message
if 'debug_messages' in st.session_state:
st.session_state.debug_messages.append(f"Executing {tool.name}")
# Call the tool
result = asyncio.run(call_tool(command, args, tool.name, tool_inputs))
# Display result
st.subheader("Result")
st.write(result)
else:
st.warning("No input schema available for this tool")
# Display resources if any
if server_info.get('resources'):
with st.expander("Resources"):
for resource in server_info['resources']:
st.write(f"**{resource.name}:** {resource.uri}")
if hasattr(resource, 'description') and resource.description:
st.write(resource.description)
st.divider()
# Display prompts if any
if server_info.get('prompts'):
with st.expander("Prompts"):
for prompt in server_info['prompts']:
st.write(f"**{prompt.name}**")
if hasattr(prompt, 'description') and prompt.description:
st.write(prompt.description)
st.divider()
else:
st.info(f"Server {active_server} is selected but not connected. Click 'Connect' in the sidebar.")
else:
# Initial state when no server is connected
st.info("Select a server from the sidebar and click 'Connect' to start interacting with it.")