import streamlit as st
import asyncio
import sys
import os
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
# Page Config
st.set_page_config(
page_title="Open Food Facts MCP Inspector",
page_icon="🍫",
layout="wide"
)
# Title
st.title("🍫 Open Food Facts MCP Inspector")
st.markdown("Interact with the Open Food Facts MCP Server locally.")
# Helper to run async code in Streamlit
def run_async(coro):
return asyncio.run(coro)
async def call_mcp_tool(tool_name, arguments):
"""
Connects to the local MCP server, calls a tool, and returns the result.
Note: In a production app, the connection might be persistent or managed differently.
Here strictly for verifying the CLI/Stdio server.
"""
server_script = os.path.join(os.getcwd(), "server.py")
server_params = StdioServerParameters(
command=sys.executable,
args=[server_script],
env=None
)
try:
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
# Check if tool exists
tools = await session.list_tools()
tool_names = [t.name for t in tools.tools]
if tool_name not in tool_names:
return f"Error: Tool '{tool_name}' not found. Available: {tool_names}"
result = await session.call_tool(tool_name, arguments=arguments)
if result.content:
return result.content[0].text
return "No content returned."
except Exception as e:
return f"Error connecting to server: {str(e)}"
async def get_tools_list():
server_script = os.path.join(os.getcwd(), "server.py")
server_params = StdioServerParameters(
command=sys.executable,
args=[server_script],
env=None
)
try:
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
tools = await session.list_tools()
return tools.tools
except Exception as e:
return []
# Sidebar
st.sidebar.header("Server Status")
st.sidebar.success("Server Script Detected: server.py")
if st.sidebar.button("Check Available Tools"):
with st.spinner("Connecting to MCP Server..."):
tools = run_async(get_tools_list())
if tools:
st.sidebar.markdown("### Available Tools:")
for t in tools:
st.sidebar.text(f"• {t.name}")
else:
st.sidebar.error("Could not fetch tools. Is server.py valid?")
# Main Interface
tab1, tab2 = st.tabs(["🔍 Search Products", "🏷️ Product Details"])
with tab1:
st.header("Search Products")
col1, col2 = st.columns([3, 1])
with col1:
search_query = st.text_input("Enter product name (e.g., 'Nutella', 'Coca Cola')", key="search_q")
with col2:
search_btn = st.button("Search", key="search_btn")
if search_btn and search_query:
with st.spinner(f"Searching for '{search_query}'..."):
result = run_async(call_mcp_tool("search_products", {"query": search_query}))
if "Error" in result:
st.error(result)
else:
st.success("Search Complete")
st.markdown(result)
with tab2:
st.header("Get Product Details")
col1, col2 = st.columns([3, 1])
with col1:
barcode_input = st.text_input("Enter Barcode (e.g., 3017620422003)", key="barcode_in")
with col2:
details_btn = st.button("Get Details", key="details_btn")
if details_btn and barcode_input:
with st.spinner(f"Fetching details for {barcode_input}..."):
result = run_async(call_mcp_tool("get_product_by_barcode", {"barcode": barcode_input}))
if "Error" in result:
st.error(result)
else:
st.success("Product Found")
st.code(result, language="text")
st.markdown("---")
st.caption("Powered by Open Food Facts & Model Context Protocol")
