Spaces:
Sleeping
Sleeping
| import gradio as gr | |
| import os | |
| import json | |
| from typing import List, Dict, Any | |
| try: | |
| from smolagents import InferenceClientModel, CodeAgent, MCPClient | |
| HAS_SMOLAGENTS = True | |
| except ImportError: | |
| HAS_SMOLAGENTS = False | |
| class SimpleMCPClient: | |
| """Simple MCP Client for Hugging Face Spaces""" | |
| def __init__(self): | |
| self.current_client = None | |
| self.current_agent = None | |
| self.conversation_history = [] | |
| # Default MCP server | |
| self.default_server_url = "https://abidlabs-mcp-tool-http.hf.space/gradio_api/mcp/sse" | |
| def connect_to_server(self, server_url: str = None) -> tuple[bool, str]: | |
| """Connect to MCP server""" | |
| url = server_url or self.default_server_url | |
| try: | |
| if HAS_SMOLAGENTS: | |
| # Disconnect existing connection | |
| if self.current_client: | |
| try: | |
| self.current_client.disconnect() | |
| except: | |
| pass | |
| # Create new connection | |
| self.current_client = MCPClient({"url": url}) | |
| tools = self.current_client.get_tools() | |
| # Create agent | |
| model = InferenceClientModel(token=os.getenv("HF_TOKEN")) | |
| self.current_agent = CodeAgent( | |
| tools=[*tools], | |
| model=model, | |
| additional_authorized_imports=["json", "ast", "urllib", "base64"] | |
| ) | |
| return True, f"✅ Connected successfully! Available tools: {len(tools)}" | |
| else: | |
| # Fallback for when smolagents is not available | |
| return True, "✅ Connected (simulated - smolagents not available)" | |
| except Exception as e: | |
| return False, f"❌ Connection failed: {str(e)}" | |
| def chat_with_agent(self, message: str) -> str: | |
| """Chat with the connected agent""" | |
| if not message.strip(): | |
| return "❌ Please enter a message." | |
| if not self.current_agent and HAS_SMOLAGENTS: | |
| return "❌ Please connect to a server first." | |
| try: | |
| if HAS_SMOLAGENTS and self.current_agent: | |
| response = str(self.current_agent.run(message)) | |
| else: | |
| # Fallback response | |
| response = f"Simulated response: I received your message '{message}' but cannot process it without smolagents." | |
| # Add to history | |
| self.conversation_history.append({ | |
| "user": message, | |
| "assistant": response | |
| }) | |
| return response | |
| except Exception as e: | |
| error_msg = f"❌ Error: {str(e)}" | |
| self.conversation_history.append({ | |
| "user": message, | |
| "assistant": error_msg | |
| }) | |
| return error_msg | |
| def get_conversation_history(self) -> List[List[str]]: | |
| """Get conversation history in Gradio ChatInterface format""" | |
| history = [] | |
| for conv in self.conversation_history: | |
| history.append([conv["user"], conv["assistant"]]) | |
| return history | |
| def disconnect(self): | |
| """Disconnect from server""" | |
| if self.current_client and HAS_SMOLAGENTS: | |
| try: | |
| self.current_client.disconnect() | |
| except: | |
| pass | |
| self.current_client = None | |
| self.current_agent = None | |
| # Global client instance | |
| simple_client = SimpleMCPClient() | |
| # Gradio Interface Functions | |
| def connect_wrapper(server_url: str) -> str: | |
| """Wrapper for connection""" | |
| success, message = simple_client.connect_to_server(server_url) | |
| return message | |
| def chat_wrapper(message: str, history: List[List[str]]) -> str: | |
| """Wrapper for chat""" | |
| return simple_client.chat_with_agent(message) | |
| # Pre-defined servers for easy selection | |
| SERVER_OPTIONS = { | |
| "MCP Tools Collection": "https://abidlabs-mcp-tool-http.hf.space/gradio_api/mcp/sse", | |
| # "Sentiment Analyzer": "https://huggingface.co/spaces/Adilbai/MCP-sentiment-analyzer/gradio_api/mcp/sse", # Removed sentiment analyzer | |
| "Custom": "" | |
| } | |
| # Example prompts | |
| EXAMPLE_PROMPTS = [ | |
| "What is the prime factorization of 68?", | |
| "Calculate the square root of 144", | |
| # "Analyze the sentiment of: 'I love this product!'", # Removed sentiment example | |
| "Convert 'Hello World' to base64", | |
| "What tools are available?", | |
| "Help me with a math problem" | |
| ] | |
| # Create Gradio Interface | |
| with gr.Blocks( | |
| theme=gr.themes.Soft(), | |
| title="🌐 Simple MCP Client", | |
| css=""" | |
| .gradio-container { | |
| max-width: 1200px !important; | |
| margin: auto !important; | |
| } | |
| .main-header { | |
| text-align: center; | |
| background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); | |
| -webkit-background-clip: text; | |
| -webkit-text-fill-color: transparent; | |
| background-clip: text; | |
| font-size: 2.5em; | |
| font-weight: bold; | |
| margin-bottom: 20px; | |
| } | |
| """ | |
| ) as demo: | |
| gr.HTML(""" | |
| <div class="main-header"> | |
| 🌐 Simple MCP Client | |
| </div> | |
| <p style="text-align: center; font-size: 1.2em; color: #666; margin-bottom: 30px;"> | |
| Connect to MCP servers and chat with AI agents | |
| </p> | |
| """) | |
| with gr.Tabs(): | |
| # Main Chat Tab | |
| with gr.Tab("💬 Agent Chat"): | |
| with gr.Row(): | |
| with gr.Column(scale=2): | |
| # Chat interface | |
| chatbot = gr.Chatbot( | |
| label="🤖 MCP Agent Chat", | |
| height=400, | |
| show_label=True | |
| ) | |
| msg = gr.Textbox( | |
| label="Your Message", | |
| placeholder="Type your message here...", | |
| lines=2, | |
| max_lines=5 | |
| ) | |
| with gr.Row(): | |
| send_btn = gr.Button("📤 Send", variant="primary", size="lg") | |
| clear_btn = gr.Button("🗑️ Clear Chat", variant="secondary") | |
| with gr.Column(scale=1): | |
| gr.Markdown("### 🔌 Server Connection") | |
| server_select = gr.Dropdown( | |
| choices=list(SERVER_OPTIONS.keys()), | |
| value="MCP Tools Collection" if "MCP Tools Collection" in SERVER_OPTIONS else (list(SERVER_OPTIONS.keys())[0] if SERVER_OPTIONS else "Custom"), # Adjusted default value | |
| label="Select Server" | |
| ) | |
| custom_url = gr.Textbox( | |
| label="Custom Server URL", | |
| placeholder="https://your-server.com/gradio_api/mcp/sse", | |
| visible=(server_select.value == "Custom") # Adjusted visibility based on initial value | |
| ) | |
| connect_btn = gr.Button("🔌 Connect to Server", variant="primary") | |
| connection_status = gr.Textbox( | |
| label="Connection Status", | |
| value="Not connected", | |
| interactive=False, | |
| lines=3 | |
| ) | |
| gr.Markdown("### 🎯 Example Prompts") | |
| example_dropdown = gr.Dropdown( | |
| choices=EXAMPLE_PROMPTS, | |
| label="Select an example", | |
| value=None | |
| ) | |
| # Server Management Tab | |
| with gr.Tab("🖥️ Server Management"): | |
| gr.Markdown("## 🔧 MCP Server Configuration") | |
| with gr.Row(): | |
| with gr.Column(): | |
| gr.Markdown("### 📋 Available Servers") | |
| for name, url in SERVER_OPTIONS.items(): | |
| if name != "Custom": | |
| gr.Markdown(f""" | |
| **{name}** | |
| - URL: `{url}` | |
| - Status: {'🟢 Available' if url else '🔴 Not configured'} | |
| """) | |
| with gr.Column(): | |
| gr.Markdown("### ➕ Add Custom Server") | |
| custom_name = gr.Textbox( | |
| label="Server Name", | |
| placeholder="My Custom Server" | |
| ) | |
| custom_server_url = gr.Textbox( | |
| label="Server URL", | |
| placeholder="https://your-mcp-server.com/gradio_api/mcp/sse" | |
| ) | |
| test_connection_btn = gr.Button("🧪 Test Connection", variant="secondary") | |
| test_result = gr.Textbox( | |
| label="Test Result", | |
| interactive=False, | |
| lines=2 | |
| ) | |
| # Help & About Tab | |
| with gr.Tab("ℹ️ Help & About"): | |
| gr.Markdown(""" | |
| ## 🌐 Simple MCP Client | |
| This is a streamlined **Model Context Protocol (MCP) Client** for connecting to MCP servers and interacting with AI agents. | |
| ### 🚀 Quick Start | |
| 1. **Connect to a Server:** | |
| - Select a pre-configured server from the dropdown | |
| - Click "Connect to Server" | |
| - Wait for connection confirmation | |
| 2. **Start Chatting:** | |
| - Type your message in the text box | |
| - Click "Send" or press Enter | |
| - The AI agent will respond using available MCP tools | |
| 3. **Try Examples:** | |
| - Use the "Example Prompts" dropdown for inspiration | |
| - Different servers have different capabilities | |
| ### 🔧 Pre-configured Servers | |
| **MCP Tools Collection:** General-purpose tools including math, text processing, and utilities | |
| ### 🎯 Example Use Cases | |
| - **Math Calculations:** "What is the prime factorization of 68?" | |
| - **Text Processing:** "Convert 'Hello World' to base64" | |
| - **Data Analysis:** "Calculate statistics for [1,2,3,4,5]" | |
| ### 🔧 Technical Requirements | |
| - **smolagents:** For full MCP client functionality | |
| - **Hugging Face Token:** Set `HF_TOKEN` environment variable | |
| - **Internet Connection:** Required for server communication | |
| ### 🆘 Troubleshooting | |
| **Connection Issues:** | |
| - Verify server URL is correct | |
| - Check internet connectivity | |
| - Ensure Hugging Face token is valid | |
| **Chat Issues:** | |
| - Make sure you're connected to a server | |
| - Try reconnecting if responses are slow | |
| - Check if the server supports your request type | |
| --- | |
| **Current Status:** | |
| - smolagents Available: '✅ Yes' | |
| - MCP Protocol Version: 2024-11-05 | |
| - Client Version: 1.0.0 | |
| """) | |
| # Event Handlers | |
| # Show/hide custom URL input based on server selection | |
| def update_custom_url_visibility(server_name): | |
| return gr.update(visible=(server_name == "Custom")) | |
| server_select.change( | |
| fn=update_custom_url_visibility, | |
| inputs=server_select, | |
| outputs=custom_url | |
| ) | |
| # Connect to server | |
| def connect_to_selected_server(server_name, custom_url_value): | |
| if server_name == "Custom": | |
| url = custom_url_value | |
| else: | |
| url = SERVER_OPTIONS.get(server_name) | |
| if not url: | |
| return "❌ Please provide a server URL" | |
| return connect_wrapper(url) | |
| connect_btn.click( | |
| fn=connect_to_selected_server, | |
| inputs=[server_select, custom_url], | |
| outputs=connection_status | |
| ) | |
| # Chat functionality | |
| def chat_function(message, history): | |
| if not message.strip(): | |
| return history, "" # Return current history and empty message box | |
| response = simple_client.chat_with_agent(message) | |
| # The history is managed by Gradio's Chatbot by returning a list of lists | |
| # We need to append to the history list passed in and return it | |
| new_history = history + [[message, response]] | |
| return new_history, "" # Return updated history and clear message box | |
| send_btn.click( | |
| fn=chat_function, | |
| inputs=[msg, chatbot], | |
| outputs=[chatbot, msg] | |
| ) | |
| msg.submit( | |
| fn=chat_function, | |
| inputs=[msg, chatbot], | |
| outputs=[chatbot, msg] | |
| ) | |
| # Clear chat | |
| def clear_chat_history(): | |
| simple_client.conversation_history = [] # Clear internal history | |
| return [], "" # Clear chatbot display and message input | |
| clear_btn.click( | |
| fn=clear_chat_history, | |
| inputs=None, # No inputs needed | |
| outputs=[chatbot, msg] # Clear chatbot and message box | |
| ) | |
| # Example prompts | |
| example_dropdown.change( | |
| fn=lambda x: x if x else "", | |
| inputs=example_dropdown, | |
| outputs=msg | |
| ) | |
| # Test connection | |
| def test_custom_connection(url): | |
| if not url.strip(): | |
| return "❌ Please enter a URL to test" | |
| # Temporarily connect and disconnect to test | |
| original_client = simple_client.current_client | |
| original_agent = simple_client.current_agent | |
| success, message = simple_client.connect_to_server(url) | |
| # Restore original state if it was connected | |
| if original_client: | |
| simple_client.current_client = original_client | |
| simple_client.current_agent = original_agent | |
| else: # if it wasn't connected before, disconnect the test connection | |
| simple_client.disconnect() | |
| return message | |
| test_connection_btn.click( | |
| fn=test_custom_connection, | |
| inputs=custom_server_url, | |
| outputs=test_result | |
| ) | |
| # Auto-connect to default server on startup | |
| def auto_connect(): | |
| """Auto-connect to default server""" | |
| # Ensure default_server_url is valid after changes to SERVER_OPTIONS | |
| default_server_key = "MCP Tools Collection" # Or another default if this is removed | |
| if default_server_key in SERVER_OPTIONS: | |
| url_to_connect = SERVER_OPTIONS[default_server_key] | |
| success, message = simple_client.connect_to_server(url_to_connect) | |
| print(f"Auto-connect to '{default_server_key}': {message}") | |
| elif SERVER_OPTIONS: # Connect to the first available if default is gone | |
| first_server_key = list(SERVER_OPTIONS.keys())[0] | |
| if first_server_key != "Custom": | |
| url_to_connect = SERVER_OPTIONS[first_server_key] | |
| success, message = simple_client.connect_to_server(url_to_connect) | |
| print(f"Auto-connect to '{first_server_key}': {message}") | |
| else: | |
| print("Auto-connect: No default server available to connect.") | |
| else: | |
| print("Auto-connect: No servers configured.") | |
| # Launch configuration | |
| if __name__ == "__main__": | |
| print("🚀 Starting Simple MCP Client...") | |
| print(f"📊 smolagents available: {HAS_SMOLAGENTS}") | |
| # Update HAS_SMOLAGENTS in Help Tab dynamically | |
| help_tab_index = -1 | |
| for i, tab_item in enumerate(demo.children): | |
| if isinstance(tab_item, gr.layouts.Tabs): | |
| for j, tab_child in enumerate(tab_item.children): | |
| if isinstance(tab_child, gr.layouts.Tab) and tab_child.label == "ℹ️ Help & About": | |
| help_tab_index = (i,j) | |
| break | |
| if help_tab_index != -1: | |
| break | |
| if help_tab_index != -1: | |
| pass | |
| print("🔌 Auto-connecting to a server...") | |
| auto_connect() | |
| # Determine if running on HF Spaces | |
| is_hf_space = os.getenv("SPACE_ID") is not None | |
| if is_hf_space: | |
| print("🚀 Running on Hugging Face Spaces") | |
| demo.launch() | |
| else: | |
| print("🌐 Web UI: http://localhost:7860") | |
| demo.launch( | |
| server_name="127.0.0.1", | |
| server_port=7860, | |
| show_error=True, | |
| inbrowser=True, | |
| debug=True | |
| ) |