app.py

import gradio as gr
import requests
import json
from openfloor import (
    Envelope, Conversation, Sender, DialogEvent, TextFeature, 
    UtteranceEvent, GetManifestsEvent, To
)

# Function to build the envelope for a manifest request using Open Floor library
def build_manifest_request_envelope(agent_url):
    # Create conversation and sender
    conversation = Conversation()
    sender = Sender(
        speakerUri="openfloor://localhost/TestClient",
        serviceUrl="http://localhost"
    )
    
    # Create envelope
    envelope = Envelope(conversation=conversation, sender=sender)
    
    # Add get manifests event
    manifest_event = GetManifestsEvent(
        to=To(serviceUrl=agent_url, private=False)
    )
    envelope.events.append(manifest_event)
    
    return envelope

# Function to build an utterance envelope using Open Floor library
def build_utterance_envelope(agent_url, message_text):
    # Create conversation and sender
    conversation = Conversation()
    sender = Sender(
        speakerUri="openfloor://localhost/TestClient",
        serviceUrl="http://localhost"
    )
    
    # Create envelope
    envelope = Envelope(conversation=conversation, sender=sender)
    
    # Create dialog event with text feature
    dialog_event = DialogEvent(
        speakerUri="openfloor://localhost/TestClient",
        features={"text": TextFeature(values=[message_text])}
    )
    
    # Create utterance event
    utterance_event = UtteranceEvent(
        dialogEvent=dialog_event,
        to=To(serviceUrl=agent_url, private=False)
    )
    
    envelope.events.append(utterance_event)
    
    return envelope

# Function to send the request to the agent
def send_request(agent_url, request_type, message_text="Hello, this is a test message!"):
    try:
        if request_type == "manifest":
            envelope = build_manifest_request_envelope(agent_url)
        else:  # utterance
            envelope = build_utterance_envelope(agent_url, message_text)
        
        # Convert to JSON using the library's method
        json_str = envelope.to_json()
        envelope_dict = json.loads(json_str)
        json_payload_pretty = json.dumps({"openFloor": envelope_dict}, indent=2)
        
        # Send request
        headers = {'Content-Type': 'application/json'}
        payload = {"openFloor": envelope_dict}
        response = requests.post(agent_url, json=payload, headers=headers, timeout=10)
        
        # Handle response
        try:
            response_json = response.json()
        except:
            response_json = {"raw_response": response.text}
        
        # Return: envelope, status, response, status_message
        status_message = f"✅ Request sent successfully (HTTP {response.status_code})"
        return json_payload_pretty, response.status_code, response_json, status_message
        
    except Exception as e:
        error_response = {"error": f"Error: {str(e)}"}
        return "", 500, error_response, f"❌ Request failed: {str(e)}"

# Function to test agent availability
def test_agent_availability(agent_url):
    try:
        # Send a simple manifest request to test
        envelope = build_manifest_request_envelope(agent_url)
        json_str = envelope.to_json()
        envelope_dict = json.loads(json_str)
        payload = {"openFloor": envelope_dict}
        
        headers = {'Content-Type': 'application/json'}
        response = requests.post(agent_url, json=payload, headers=headers, timeout=5)
        
        if response.status_code == 200:
            try:
                response_json = response.json()
                return f"✅ Agent available at {agent_url}", response_json
            except:
                return f"✅ Agent responded but with non-JSON: {response.text[:100]}...", {"raw_response": response.text}
        else:
            return f"⚠️ Agent responded with status {response.status_code}", {"status": response.status_code, "response": response.text}
            
    except requests.exceptions.RequestException as e:
        return f"❌ Agent not available: {str(e)}", {"error": str(e)}

# Enhanced test function that also shows the envelope
def test_agent_with_envelope(agent_url):
    try:
        # Send a simple manifest request to test
        envelope = build_manifest_request_envelope(agent_url)
        json_str = envelope.to_json()
        envelope_dict = json.loads(json_str)
        json_payload_pretty = json.dumps({"openFloor": envelope_dict}, indent=2)
        payload = {"openFloor": envelope_dict}
        
        headers = {'Content-Type': 'application/json'}
        response = requests.post(agent_url, json=payload, headers=headers, timeout=5)
        
        if response.status_code == 200:
            try:
                response_json = response.json()
                status_msg = f"✅ Agent available at {agent_url}"
                return status_msg, response_json, json_payload_pretty, response.status_code
            except:
                status_msg = f"✅ Agent responded but with non-JSON"
                return status_msg, {"raw_response": response.text}, json_payload_pretty, response.status_code
        else:
            status_msg = f"⚠️ Agent responded with status {response.status_code}"
            return status_msg, {"status": response.status_code, "response": response.text}, json_payload_pretty, response.status_code
            
    except requests.exceptions.RequestException as e:
        status_msg = f"❌ Agent not available: {str(e)}"
        return status_msg, {"error": str(e)}, "", 0

# Gradio UI with improved layout
with gr.Blocks(title="Open Floor Protocol Tester", theme=gr.themes.Soft()) as demo:
    gr.Markdown("""
    # 🗣️ Open Floor Protocol Tester
    **Test [Open Floor](https://github.com/open-voice-interoperability/openfloor-docs)-compliant assistants using the [official Python library](https://github.com/open-voice-interoperability/openfloor-python)**
    
    Send **manifest** requests to get capabilities or **utterance** events for conversations.
    """)
    
    with gr.Row():
        # Left Column - Controls
        with gr.Column(scale=2):
            gr.Markdown("### 🎯 Assistant Configuration")
            
            agent_url = gr.Textbox(
                value="", 
                label="Assistant Endpoint URL",
                placeholder="https://your-assistant.com/openfloor",
                info="Full URL endpoint of the Open Floor assistant"
            )
            
            gr.Markdown("### 📨 Send Request")
            
            request_type = gr.Radio(
                choices=["manifest", "utterance"],
                value="manifest",
                label="Request Type",
                info="manifest = test availability + get capabilities, utterance = send message"
            )
            
            message_text = gr.Textbox(
                value="convert 20 millimeters to gallons",
                label="Message Text",
                placeholder="Enter your message here...",
                lines=3,
                info="Text to send in utterance event"
            )
            
            send_btn = gr.Button("🚀 Send Open Floor Request", variant="primary", size="lg")
        
        # Right Column - Results  
        with gr.Column(scale=3):
            gr.Markdown("### 📡 Request & Response")
            
            # Status message for all requests
            request_status = gr.Textbox(
                label="Status",
                interactive=False,
                container=False
            )
            
            # Collapsible request details
            with gr.Accordion("📤 Generated Open Floor Envelope", open=False):
                envelope_output = gr.Code(
                    language="json", 
                    label="Request JSON",
                    container=False
                )
            
            # Response section
            with gr.Group():
                gr.Markdown("#### 📥 Response")
                
                with gr.Row():
                    status_code = gr.Number(
                        label="HTTP Status", 
                        interactive=False,
                        container=False,
                        scale=1
                    )
                
                response_output = gr.JSON(
                    label="Response Data",
                    container=False
                )
    
    # Event handlers
    def handle_send_request(*args):
        envelope, status, response, status_msg = send_request(*args)
        return envelope, status, response, status_msg
    
    send_btn.click(
        fn=handle_send_request,
        inputs=[agent_url, request_type, message_text],
        outputs=[envelope_output, status_code, response_output, request_status]
    )

if __name__ == "__main__":
    demo.launch()