Create app.py

app.py

@@ -0,0 +1,568 @@
+import gradio as gr
+import requests
+import json
+import uuid
+import threading
+from datetime import datetime
+from typing import List, Dict, Any, Optional
+from openfloor import (
+    Envelope, Conversation, Sender, DialogEvent, TextFeature, 
+    UtteranceEvent, GetManifestsEvent, To
+)
+from openfloor.manifest import *
+from openfloor.envelope import *
+
+class OpenFloorAgent:
+    def __init__(self, url: str, name: str = None):
+        self.url = url
+        self.name = name or url
+        self.manifest = None
+        self.capabilities = []
+    
+    def get_manifest(self):
+        """Get the agent's manifest and capabilities"""
+        try:
+            # Create manifest request
+            conversation = Conversation()
+            sender = Sender(
+                speakerUri="openfloor://mcp-bridge/client",
+                serviceUrl="http://mcp-bridge"
+            )
+            envelope = Envelope(conversation=conversation, sender=sender)
+            
+            manifest_event = GetManifestsEvent(
+                to=To(serviceUrl=self.url, private=False)
+            )
+            envelope.events.append(manifest_event)
+            
+            # Send request
+            json_str = envelope.to_json()
+            envelope_dict = json.loads(json_str)
+            payload = {"openFloor": envelope_dict}
+            
+            response = requests.post(self.url, json=payload, timeout=10)
+            
+            if response.status_code == 200:
+                response_data = response.json()
+                # Extract manifest from response
+                if "openFloor" in response_data and "events" in response_data["openFloor"]:
+                    for event in response_data["openFloor"]["events"]:
+                        if event.get("eventType") == "publishManifests":
+                            manifests = event.get("parameters", {}).get("manifests", [])
+                            if manifests:
+                                self.manifest = manifests[0]  # Take first manifest
+                                self.capabilities = self.manifest.get("capabilities", [])
+                                return True
+            return False
+        except Exception as e:
+            print(f"Error getting manifest from {self.url}: {e}")
+            return False
+    
+    def send_utterance(self, message: str) -> str:
+        """Send an utterance to the agent and get response"""
+        try:
+            # Create utterance request
+            conversation = Conversation()
+            sender = Sender(
+                speakerUri="openfloor://mcp-bridge/client",
+                serviceUrl="http://mcp-bridge"
+            )
+            envelope = Envelope(conversation=conversation, sender=sender)
+            
+            # Create dialog event
+            dialog_event = DialogEvent(
+                speakerUri="openfloor://mcp-bridge/client",
+                features={"text": TextFeature(values=[message])}
+            )
+            
+            utterance_event = UtteranceEvent(
+                dialogEvent=dialog_event,
+                to=To(serviceUrl=self.url, private=False)
+            )
+            
+            envelope.events.append(utterance_event)
+            
+            # Send request
+            json_str = envelope.to_json()
+            envelope_dict = json.loads(json_str)
+            payload = {"openFloor": envelope_dict}
+            
+            response = requests.post(self.url, json=payload, timeout=30)
+            
+            if response.status_code == 200:
+                response_data = response.json()
+                # Extract response text
+                if "openFloor" in response_data and "events" in response_data["openFloor"]:
+                    for event in response_data["openFloor"]["events"]:
+                        if event.get("eventType") == "utterance":
+                            dialog_event = event.get("parameters", {}).get("dialogEvent", {})
+                            features = dialog_event.get("features", {})
+                            text_feature = features.get("text", {})
+                            tokens = text_feature.get("tokens", [])
+                            if tokens:
+                                return tokens[0].get("value", "No response")
+                
+                return f"Agent responded but no text found: {response_data}"
+            else:
+                return f"Agent error: HTTP {response.status_code}"
+                
+        except Exception as e:
+            return f"Error communicating with agent: {e}"
+
+class BuiltInFloorManager:
+    """Built-in Floor Manager for smart agent routing"""
+    
+    def __init__(self):
+        self.agent_registry = {}  # speakerUri -> agent info
+        self.active_conversations = {}
+        
+    def register_agent_from_openfloor_agent(self, openfloor_agent: OpenFloorAgent):
+        """Register an OpenFloorAgent with the floor manager"""
+        if not openfloor_agent.manifest:
+            return False
+            
+        speaker_uri = openfloor_agent.manifest.get("identification", {}).get("speakerUri")
+        if not speaker_uri:
+            # Create a speaker URI from URL
+            speaker_uri = f"tag:{openfloor_agent.url.replace('https://', '').replace('http://', '')},2025:agent"
+            
+        self.agent_registry[speaker_uri] = {
+            'openfloor_agent': openfloor_agent,
+            'manifest': openfloor_agent.manifest,
+            'url': openfloor_agent.url,
+            'capabilities': openfloor_agent.capabilities,
+            'status': 'available',
+            'last_seen': datetime.now()
+        }
+        print(f"🏛️ Floor Manager: Registered {openfloor_agent.name}")
+        return True
+    
+    def find_best_agent_for_task(self, task: str) -> Optional[OpenFloorAgent]:
+        """Find the best agent for a given task based on capabilities"""
+        if not self.agent_registry:
+            return None
+            
+        task_lower = task.lower()
+        best_agent_info = None
+        best_score = 0
+        
+        for speaker_uri, agent_info in self.agent_registry.items():
+            score = 0
+            capabilities = agent_info.get('capabilities', [])
+            
+            for capability in capabilities:
+                keyphrases = capability.get("keyphrases", [])
+                descriptions = capability.get("descriptions", [])
+                
+                # Score based on keyphrases
+                for keyphrase in keyphrases:
+                    if keyphrase.lower() in task_lower:
+                        score += 2  # Higher weight for keyphrase matches
+                
+                # Score based on descriptions
+                for description in descriptions:
+                    if any(word in task_lower for word in description.lower().split()):
+                        score += 1
+            
+            if score > best_score:
+                best_score = score
+                best_agent_info = agent_info
+        
+        # If no capability match, use first agent
+        if not best_agent_info and self.agent_registry:
+            best_agent_info = list(self.agent_registry.values())[0]
+            
+        return best_agent_info['openfloor_agent'] if best_agent_info else None
+    
+    def get_all_capabilities(self) -> Dict[str, List[Dict]]:
+        """Get all capabilities from all registered agents"""
+        all_capabilities = {}
+        for speaker_uri, agent_info in self.agent_registry.items():
+            agent_name = agent_info['openfloor_agent'].name
+            all_capabilities[agent_name] = agent_info.get('capabilities', [])
+        return all_capabilities
+    
+    def find_agents_with_capability(self, capability_keywords: List[str]) -> List[OpenFloorAgent]:
+        """Find all agents that match given capability keywords"""
+        matching_agents = []
+        keywords_lower = [kw.lower() for kw in capability_keywords]
+        
+        for speaker_uri, agent_info in self.agent_registry.items():
+            capabilities = agent_info.get('capabilities', [])
+            
+            for capability in capabilities:
+                keyphrases = capability.get("keyphrases", [])
+                descriptions = capability.get("descriptions", [])
+                
+                # Check if any keyword matches keyphrases or descriptions
+                for keyphrase in keyphrases:
+                    if any(kw in keyphrase.lower() for kw in keywords_lower):
+                        matching_agents.append(agent_info['openfloor_agent'])
+                        break
+                
+                if agent_info['openfloor_agent'] in matching_agents:
+                    break
+                    
+                for description in descriptions:
+                    if any(kw in description.lower() for kw in keywords_lower):
+                        matching_agents.append(agent_info['openfloor_agent'])
+                        break
+                
+                if agent_info['openfloor_agent'] in matching_agents:
+                    break
+        
+        return matching_agents
+
+# Global instances
+_agents_cache: Dict[str, OpenFloorAgent] = {}
+_floor_manager = BuiltInFloorManager()
+
+def _discover_agents_from_headers(request: gr.Request) -> Dict[str, OpenFloorAgent]:
+    """Helper function to discover agents from request headers and register with floor manager"""
+    headers = dict(request.headers)
+    
+    # Support comma-separated format: x-openfloor-agents: url1,url2,url3
+    agent_urls = []
+    if "x-openfloor-agents" in headers:
+        agent_urls = [url.strip() for url in headers["x-openfloor-agents"].split(",")]
+    
+    # Update agents cache and register with floor manager
+    for url in agent_urls:
+        if url and url not in _agents_cache:
+            agent = OpenFloorAgent(url)
+            if agent.get_manifest():
+                _agents_cache[url] = agent
+                _floor_manager.register_agent_from_openfloor_agent(agent)
+                print(f"✅ Discovered and registered agent: {agent.name} at {url}")
+            else:
+                print(f"❌ Failed to get manifest from: {url}")
+    
+    return _agents_cache
+
+def discover_openfloor_agents(request: gr.Request) -> str:
+    """
+    Discover OpenFloor agents from request headers and return their capabilities.
+    
+    Provide agent URLs in the 'x-openfloor-agents' header as comma-separated values.
+    Example: x-openfloor-agents: https://agent1.com,https://agent2.com
+    
+    Args:
+        request: The HTTP request containing agent URLs in headers
+    
+    Returns:
+        str: List of discovered agents and their capabilities
+    """
+    agents = _discover_agents_from_headers(request)
+    
+    if not agents:
+        return "❌ No agents discovered. Please provide agent URLs in 'x-openfloor-agents' header as comma-separated values."
+    
+    result = "🤖 **Discovered OpenFloor Agents:**\n\n"
+    for url, agent in agents.items():
+        result += f"**{agent.name}**\n"
+        result += f"- URL: {url}\n"
+        if agent.manifest:
+            identification = agent.manifest.get("identification", {})
+            result += f"- Role: {identification.get('role', 'Unknown')}\n"
+            result += f"- Synopsis: {identification.get('synopsis', 'No description')}\n"
+        
+        result += f"- Capabilities: {len(agent.capabilities)}\n"
+        for cap in agent.capabilities:
+            keyphrases = cap.get("keyphrases", [])
+            descriptions = cap.get("descriptions", [])
+            if keyphrases:
+                result += f"  • Keywords: {', '.join(keyphrases)}\n"
+            if descriptions:
+                result += f"  • Description: {', '.join(descriptions)}\n"
+        result += "\n"
+    
+    return result
+
+def send_message_to_openfloor_agent(agent_url: str, message: str, request: gr.Request) -> str:
+    """
+    Send a message to a specific OpenFloor agent.
+    
+    Args:
+        agent_url (str): The URL of the OpenFloor agent
+        message (str): The message to send to the agent
+        request: The HTTP request (used to discover agents if needed)
+    
+    Returns:
+        str: The agent's response
+    """
+    # Ensure agents are discovered
+    agents = _discover_agents_from_headers(request)
+    
+    if agent_url not in agents:
+        return f"❌ Agent not found: {agent_url}. Please discover agents first or check the URL."
+    
+    agent = agents[agent_url]
+    response = agent.send_utterance(message)
+    return f"🤖 **{agent.name}**: {response}"
+
+def send_to_best_openfloor_agent(task_description: str, request: gr.Request) -> str:
+    """
+    Send a task to the best available OpenFloor agent using built-in floor manager.
+    
+    The built-in floor manager analyzes agent capabilities and selects the most suitable agent.
+    
+    Args:
+        task_description (str): Description of the task to be performed
+        request: The HTTP request (used to discover agents)
+    
+    Returns:
+        str: The selected agent's response with agent selection details
+    """
+    # Ensure agents are discovered and registered with floor manager
+    _discover_agents_from_headers(request)
+    
+    # Use built-in floor manager to find best agent
+    best_agent = _floor_manager.find_best_agent_for_task(task_description)
+    
+    if not best_agent:
+        return "❌ No agents available. Please provide 'x-openfloor-agents' header with valid agent URLs."
+    
+    # Send message to selected agent
+    response = best_agent.send_utterance(task_description)
+    
+    return f"🏛️ **Floor Manager Selected: {best_agent.name}**\n\n🤖 **Response**: {response}"
+
+def execute_agent_capability(capability_keywords: str, task_request: str, request: gr.Request) -> str:
+    """
+    Execute a task using agents that have specific capabilities.
+    
+    This function finds agents based on their actual manifest capabilities and sends the task.
+    
+    Args:
+        capability_keywords (str): Comma-separated keywords to match against agent capabilities (e.g., "convert,unit,calculation")
+        task_request (str): The actual task to send to the matching agent
+        request: The HTTP request (used to discover agents)
+    
+    Returns:
+        str: The response from the matching agent
+    """
+    # Ensure agents are discovered
+    _discover_agents_from_headers(request)
+    
+    # Parse capability keywords
+    keywords = [kw.strip() for kw in capability_keywords.split(",")]
+    
+    # Find agents with matching capabilities
+    matching_agents = _floor_manager.find_agents_with_capability(keywords)
+    
+    if not matching_agents:
+        available_capabilities = _floor_manager.get_all_capabilities()
+        capability_summary = ""
+        for agent_name, capabilities in available_capabilities.items():
+            capability_summary += f"\n- {agent_name}: "
+            all_keywords = []
+            for cap in capabilities:
+                all_keywords.extend(cap.get("keyphrases", []))
+            capability_summary += ", ".join(all_keywords[:5])
+        
+        return f"❌ No agents found with capabilities matching: {capability_keywords}\n\nAvailable capabilities:{capability_summary}"
+    
+    # Use the first matching agent (or could implement scoring here)
+    selected_agent = matching_agents[0]
+    response = selected_agent.send_utterance(task_request)
+    
+    return f"🎯 **Selected Agent: {selected_agent.name}** (matched: {capability_keywords})\n\n🤖 **Response**: {response}"
+
+def list_all_agent_capabilities(request: gr.Request) -> str:
+    """
+    List all capabilities of all discovered OpenFloor agents.
+    
+    This shows what the agents can actually do according to their manifests.
+    
+    Args:
+        request: The HTTP request (used to discover agents)
+    
+    Returns:
+        str: Comprehensive list of all agent capabilities
+    """
+    # Ensure agents are discovered
+    _discover_agents_from_headers(request)
+    
+    all_capabilities = _floor_manager.get_all_capabilities()
+    
+    if not all_capabilities:
+        return "❌ No agents discovered. Please provide 'x-openfloor-agents' header."
+    
+    result = "🎯 **All Agent Capabilities:**\n\n"
+    
+    for agent_name, capabilities in all_capabilities.items():
+        result += f"**{agent_name}:**\n"
+        
+        if not capabilities:
+            result += "  - No capabilities defined\n\n"
+            continue
+            
+        for i, capability in enumerate(capabilities, 1):
+            keyphrases = capability.get("keyphrases", [])
+            descriptions = capability.get("descriptions", [])
+            languages = capability.get("languages", [])
+            
+            result += f"  **Capability {i}:**\n"
+            if keyphrases:
+                result += f"    • Keywords: {', '.join(keyphrases)}\n"
+            if descriptions:
+                result += f"    • Can do: {', '.join(descriptions)}\n"
+            if languages:
+                result += f"    • Languages: {', '.join(languages)}\n"
+            result += "\n"
+        
+        result += "\n"
+    
+    return result
+
+def send_task_to_agents_with_keywords(keywords: str, task: str, request: gr.Request) -> str:
+    """
+    Send a task to all agents that match the given capability keywords.
+    
+    Args:
+        keywords (str): Comma-separated capability keywords to search for
+        task (str): The task to send to matching agents
+        request: The HTTP request (used to discover agents)
+    
+    Returns:
+        str: Responses from all matching agents
+    """
+    # Ensure agents are discovered
+    _discover_agents_from_headers(request)
+    
+    # Parse keywords
+    keyword_list = [kw.strip() for kw in keywords.split(",")]
+    
+    # Find matching agents
+    matching_agents = _floor_manager.find_agents_with_capability(keyword_list)
+    
+    if not matching_agents:
+        return f"❌ No agents found with capabilities matching: {keywords}"
+    
+    result = f"🎯 **Found {len(matching_agents)} agents matching '{keywords}':**\n\n"
+    
+    for agent in matching_agents:
+        response = agent.send_utterance(task)
+        result += f"**{agent.name}:** {response}\n\n"
+    
+    return result
+
+# Create Gradio interface for testing
+with gr.Blocks(title="MCP-OpenFloor Bridge (Dynamic Capabilities)", theme=gr.themes.Soft()) as demo:
+    gr.Markdown("""
+    # 🌉 MCP-OpenFloor Bridge Server (Dynamic Capabilities)
+    
+    This MCP server dynamically uses actual OpenFloor agent capabilities from their manifests.
+    
+    ## 🔧 Setup Instructions:
+    
+    ```json
+    {
+      "mcpServers": {
+        "openfloor-bridge": {
+          "url": "http://localhost:7860/gradio_api/mcp/sse",
+          "headers": {
+            "x-openfloor-agents": "https://beaconforge.pythonanywhere.com,https://agent2.com"
+          }
+        }
+      }
+    }
+    ```
+    
+    ## 🎯 Dynamic Capability-Based Tools:
+    - **discover_openfloor_agents**: Find agents and show their actual capabilities
+    - **list_all_agent_capabilities**: Show what each agent can actually do
+    - **send_to_best_openfloor_agent**: Smart routing based on agent capabilities
+    - **execute_agent_capability**: Execute tasks using specific capability keywords
+    - **send_task_to_agents_with_keywords**: Send to all agents matching keywords
+    - **send_message_to_openfloor_agent**: Direct communication with specific agent
+    
+    ## 📋 No Hardcoded Functions:
+    - Tools are based on actual agent capabilities from manifests
+    - Use capability keywords to find appropriate agents
+    - Dynamic routing based on what agents can actually do
+    """)
+    
+    with gr.Row():
+        with gr.Column():
+            gr.Markdown("### 🔍 Agent Discovery")
+            discover_btn = gr.Button("Discover Agents", variant="primary")
+            agent_list = gr.Textbox(label="Discovered Agents", lines=8)
+            
+            capabilities_btn = gr.Button("List All Capabilities", variant="secondary")
+            capabilities_output = gr.Textbox(label="Agent Capabilities", lines=10)
+            
+        with gr.Column():
+            gr.Markdown("### 🎯 Capability-Based Execution")
+            keywords_input = gr.Textbox(
+                label="Capability Keywords", 
+                placeholder="convert,unit,calculation",
+                info="Comma-separated keywords to match"
+            )
+            task_input = gr.Textbox(
+                label="Task Request", 
+                placeholder="convert 50 meters to feet"
+            )
+            execute_btn = gr.Button("Execute with Matching Agent", variant="primary")
+            execute_output = gr.Textbox(label="Execution Result", lines=8)
+    
+    with gr.Row():
+        with gr.Column():
+            gr.Markdown("### 💬 Direct Agent Communication")
+            agent_url_input = gr.Textbox(
+                label="Agent URL", 
+                placeholder="https://beaconforge.pythonanywhere.com"
+            )
+            message_input = gr.Textbox(
+                label="Message", 
+                placeholder="What can you help me with?"
+            )
+            send_btn = gr.Button("Send to Specific Agent", variant="secondary")
+            response_output = gr.Textbox(label="Agent Response", lines=5)
+            
+        with gr.Column():
+            gr.Markdown("### 🏛️ Smart Floor Manager")
+            smart_task_input = gr.Textbox(
+                label="Task Description", 
+                placeholder="I need help with something"
+            )
+            smart_send_btn = gr.Button("Send to Best Agent", variant="primary")
+            smart_response = gr.Textbox(label="Floor Manager Response", lines=5)
+    
+    # Event handlers
+    discover_btn.click(
+        fn=discover_openfloor_agents,
+        inputs=[gr.Request()],
+        outputs=[agent_list]
+    )
+    
+    capabilities_btn.click(
+        fn=list_all_agent_capabilities,
+        inputs=[gr.Request()],
+        outputs=[capabilities_output]
+    )
+    
+    execute_btn.click(
+        fn=execute_agent_capability,
+        inputs=[keywords_input, task_input, gr.Request()],
+        outputs=[execute_output]
+    )
+    
+    send_btn.click(
+        fn=send_message_to_openfloor_agent,
+        inputs=[agent_url_input, message_input, gr.Request()],
+        outputs=[response_output]
+    )
+    
+    smart_send_btn.click(
+        fn=send_to_best_openfloor_agent,
+        inputs=[smart_task_input, gr.Request()],
+        outputs=[smart_response]
+    )
+
+if __name__ == "__main__":
+    demo.launch(
+        share=False,
+        debug=False,
+        mcp_server=True
+    )