File size: 5,649 Bytes
c6df9a6 556e6ef c6df9a6 556e6ef 8a3b182 c6df9a6 6cc2cc4 c6df9a6 6cc2cc4 c6df9a6 6cc2cc4 c6df9a6 8a3b182 c6df9a6 556e6ef c6df9a6 8a3b182 556e6ef c6df9a6 556e6ef 8a3b182 556e6ef c6df9a6 8a3b182 c6df9a6 6cc2cc4 c6df9a6 8a3b182 c6df9a6 6cc2cc4 c6df9a6 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 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 |
# Gradio UI for local chatbot testing with MCP client
import logging
import asyncio
import gradio as gr
from environs import Env
from mcp_client import MCPClient
logger = logging.getLogger(__name__)
env = Env()
env.read_env()
AUTH_USERNAME = env.str("AUTH_USERNAME", "admin")
AUTH_PASSWORD = env.str("AUTH_PASSWORD", "admin")
async def initialize_client():
logger.info("Initializing MCP client...")
client = MCPClient()
logger.info("MCP client created")
await client.initialize()
result = client
logger.info("MCP client initialized")
return result
def launch_ui():
with gr.Blocks(title="MCP Chatbot UI", fill_height=True, fill_width=True) as demo:
gr.Markdown("## Dnext Product Catalog AI Assistant")
# Authentication section
with gr.Row() as auth_section:
with gr.Column(scale=1):
gr.Markdown("### Authentication")
username = gr.Textbox(label="Username", placeholder="Enter your username")
password = gr.Textbox(label="Password", placeholder="Enter your password", type="password")
login_btn = gr.Button("Login", variant="primary")
# Main content section (initially hidden)
with gr.Row(visible=False) as main_content:
with gr.Column(scale=2):
gr.Markdown("### Chat Interface")
chatbot = gr.Chatbot(height=500, label="Chatbot", type="messages")
with gr.Row():
msg = gr.Textbox(
label="Enter your request",
placeholder="Type your message here...",
lines=2,
scale=4
)
submit_btn = gr.Button("Submit", scale=1)
clear_btn = gr.Button("Clear Chat", variant="secondary")
with gr.Column(scale=1):
gr.Markdown("### Example Use Cases")
with gr.Accordion("Product Creation Flow", open=False):
gr.Markdown("""
**Step 1:** Define a product characteristic
```
define a product characteristic named 'characteristic1' with various attributes like 'Simple Value', 'MBPS' as the unit of measure, and a value of '1000'
```
**Step 2:** Define price structure
```
define the price structure for the product 'price1'. The price is set to a one-time charge of 100 USD, with a unit of measure 'Day' and no recurring charges
```
**Step 3:** Create product specification
```
create a product specification for 'specification1', which links the previously created characteristic (characteristic1) to the product specification
```
**Step 4:** Define complete product offering
```
define a complete product offering. Tie the product specification (specification1), pricing (price1), and characteristics (characteristic1) that you previously created
```
""")
def authenticate(username, password):
if username == AUTH_USERNAME and password == AUTH_PASSWORD:
return {
auth_section: gr.Row.update(visible=False),
main_content: gr.Row.update(visible=True)
}
return {
auth_section: gr.Row.update(visible=True),
main_content: gr.Row.update(visible=False)
}
async def respond(user_message):
bot_response = await client.invoke(user_message)
return [
{"role": "user", "content": user_message},
{"role": "assistant", "content": bot_response},
], "" # Return empty string to clear the input field
async def clear_chat_and_memory():
await client.clear_memory()
return [], ""
login_btn.click(
fn=authenticate,
inputs=[username, password],
outputs=[auth_section, main_content]
)
submit_btn.click(fn=respond, inputs=[msg], outputs=[chatbot, msg])
clear_btn.click(fn=clear_chat_and_memory, inputs=[], outputs=[chatbot, msg])
return demo
async def main():
try:
logger.info("Starting application...")
global client
logger.info("Initializing MCP client...")
client = await initialize_client()
logger.info("MCP client initialized successfully")
logger.info("Launching Gradio UI...")
demo = launch_ui()
logger.info("Gradio UI created successfully")
logger.info(f"Starting Gradio server on {env.str('GRADIO_SERVER_HOST', '0.0.0.0')}:{env.int('GRADIO_SERVER_PORT', 7860)}")
demo.launch(
server_name=env.str("GRADIO_SERVER_HOST", "0.0.0.0"),
server_port=env.int("GRADIO_SERVER_PORT", 7860),
share=False,
show_error=True,
show_api=False
)
logger.info("Gradio server started successfully")
except Exception as e:
logger.error(f"Error starting application: {str(e)}")
raise
if __name__ == "__main__":
try:
logger.info("Starting main application...")
asyncio.run(main())
except Exception as e:
logger.error(f"Fatal error in main: {str(e)}")
raise
# uv run app.py
|