#!/usr/bin/env python3
"""
Web interface for Odoo MCP Server.
Provides a chat-like interface to interact with Odoo
and a settings page to configure connection parameters.
"""
from __future__ import annotations
import json
import os
import re
import sys
from pathlib import Path
from typing import Any
from fastapi import FastAPI, Request, Form, HTTPException
from fastapi.responses import HTMLResponse, RedirectResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from dotenv import load_dotenv, set_key
# Add parent directory to path for imports
sys.path.insert(0, str(Path(__file__).parent.parent))
# Load environment variables
ENV_FILE = Path(__file__).parent.parent / ".env"
load_dotenv(ENV_FILE)
app = FastAPI(title="Odoo MCP Web Interface", version="1.0.0")
# Mount static files and templates
STATIC_DIR = Path(__file__).parent / "static"
TEMPLATES_DIR = Path(__file__).parent / "templates"
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
templates = Jinja2Templates(directory=TEMPLATES_DIR)
def get_odoo_config() -> dict[str, str]:
"""Get current Odoo configuration from environment."""
load_dotenv(ENV_FILE, override=True)
return {
"ODOO_URL": os.getenv("ODOO_URL", ""),
"ODOO_DB": os.getenv("ODOO_DB", ""),
"ODOO_USERNAME": os.getenv("ODOO_USERNAME", ""),
"ODOO_API_KEY": os.getenv("ODOO_API_KEY", ""),
}
def save_odoo_config(config: dict[str, str]) -> None:
"""Save Odoo configuration to .env file."""
# Create .env file if it doesn't exist
if not ENV_FILE.exists():
ENV_FILE.touch()
for key, value in config.items():
set_key(str(ENV_FILE), key, value)
# Reload environment
load_dotenv(ENV_FILE, override=True)
def get_odoo_client():
"""Get a fresh Odoo client with current config."""
# Reimport to get fresh config
load_dotenv(ENV_FILE, override=True)
# Update module globals
import server
server.ODOO_URL = os.getenv("ODOO_URL", "")
server.ODOO_DB = os.getenv("ODOO_DB", "")
server.ODOO_USERNAME = os.getenv("ODOO_USERNAME", "")
server.ODOO_API_KEY = os.getenv("ODOO_API_KEY", "")
# Create new client instance
server.odoo = server.OdooClient()
return server.odoo
def markdown_to_html(text: str) -> str:
"""Convert basic markdown to HTML."""
# Headers
text = re.sub(r'^### (.+)$', r'<h4>\1</h4>', text, flags=re.MULTILINE)
text = re.sub(r'^## (.+)$', r'<h3>\1</h3>', text, flags=re.MULTILINE)
text = re.sub(r'^# (.+)$', r'<h2>\1</h2>', text, flags=re.MULTILINE)
# Bold
text = re.sub(r'\*\*(.+?)\*\*', r'<strong>\1</strong>', text)
# Italic
text = re.sub(r'\*(.+?)\*', r'<em>\1</em>', text)
# Tables
lines = text.split('\n')
in_table = False
result_lines = []
for line in lines:
if '|' in line and not line.strip().startswith('|--'):
if not in_table:
result_lines.append('<table class="result-table">')
in_table = True
# Skip separator lines
if re.match(r'^[\|\-\s]+$', line):
continue
cells = [c.strip() for c in line.split('|')[1:-1]]
if cells:
row = '<tr>' + ''.join(f'<td>{c}</td>' for c in cells) + '</tr>'
result_lines.append(row)
else:
if in_table:
result_lines.append('</table>')
in_table = False
result_lines.append(line)
if in_table:
result_lines.append('</table>')
text = '\n'.join(result_lines)
# Lists
text = re.sub(r'^- (.+)$', r'<li>\1</li>', text, flags=re.MULTILINE)
text = re.sub(r'(<li>.*</li>)', r'<ul>\1</ul>', text, flags=re.DOTALL)
# Fix nested ul
text = text.replace('</ul>\n<ul>', '\n')
# Line breaks
text = text.replace('\n\n', '</p><p>')
text = text.replace('\n', '<br>')
return f'<p>{text}</p>'
# Available commands mapping
COMMANDS = {
"projects": ("list_projects", "List all projects"),
"tasks": ("list_tasks", "List all tasks"),
"timesheets": ("list_timesheets", "List timesheet entries"),
"expenses": ("list_expenses", "List expenses"),
"contacts": ("list_contacts", "List contacts"),
"invoices": ("list_invoices", "List invoices"),
"orders": ("list_sale_orders", "List sale orders"),
"products": ("list_products", "List products"),
"employees": ("list_employees", "List employees"),
"departments": ("list_departments", "List departments"),
"categories": ("list_expense_categories", "List expense categories"),
"test": ("test_connection", "Test Odoo connection"),
"help": (None, "Show available commands"),
}
def parse_command(message: str) -> tuple[str | None, dict[str, Any]]:
"""Parse user message into command and arguments."""
message = message.strip().lower()
# Check for direct commands
parts = message.split()
if not parts:
return None, {}
cmd = parts[0]
args: dict[str, Any] = {}
# Parse key=value arguments
for part in parts[1:]:
if '=' in part:
key, value = part.split('=', 1)
# Try to convert to int/float
try:
if '.' in value:
args[key] = float(value)
else:
args[key] = int(value)
except ValueError:
args[key] = value
if cmd in COMMANDS:
return COMMANDS[cmd][0], args
# Check for search patterns
if cmd == "search":
if len(parts) >= 4:
return "search_records", {
"model": parts[1],
"field": parts[2],
"value": " ".join(parts[3:])
}
return None, {}
def execute_command(func_name: str, args: dict[str, Any]) -> str:
"""Execute an Odoo command and return the result."""
import server
# Refresh client
get_odoo_client()
if not hasattr(server, func_name):
return f"Error: Unknown function '{func_name}'"
func = getattr(server, func_name)
try:
result = func(**args)
return result
except Exception as e:
return f"Error: {str(e)}"
def get_help_text() -> str:
"""Generate help text."""
lines = ["# Available Commands\n"]
for cmd, (func, desc) in COMMANDS.items():
lines.append(f"- **{cmd}** - {desc}")
lines.append("\n## Search")
lines.append("- **search <model> <field> <value>** - Search any model")
lines.append(" - Example: `search res.partner name John`")
lines.append("\n## Arguments")
lines.append("- Add arguments with `key=value`")
lines.append(" - Example: `projects limit=10`")
lines.append(" - Example: `timesheets date_from=2025-01-01`")
return "\n".join(lines)
@app.get("/", response_class=HTMLResponse)
async def home(request: Request):
"""Render main chat interface."""
config = get_odoo_config()
is_configured = all(config.values())
return templates.TemplateResponse(
"chat.html",
{
"request": request,
"is_configured": is_configured,
"commands": COMMANDS,
}
)
@app.post("/chat")
async def chat(request: Request):
"""Handle chat messages."""
data = await request.json()
message = data.get("message", "").strip()
if not message:
return JSONResponse({"response": "Please enter a command.", "type": "error"})
# Check if help command
if message.lower() in ("help", "?", "commands"):
return JSONResponse({
"response": markdown_to_html(get_help_text()),
"type": "info"
})
# Parse command
func_name, args = parse_command(message)
if func_name is None:
return JSONResponse({
"response": f"Unknown command: '{message}'. Type <strong>help</strong> for available commands.",
"type": "error"
})
# Execute command
result = execute_command(func_name, args)
html_result = markdown_to_html(result)
result_type = "error" if result.startswith("Error") else "success"
return JSONResponse({
"response": html_result,
"type": result_type
})
@app.get("/settings", response_class=HTMLResponse)
async def settings_page(request: Request):
"""Render settings page."""
config = get_odoo_config()
return templates.TemplateResponse(
"settings.html",
{
"request": request,
"config": config,
"saved": request.query_params.get("saved") == "1",
"error": request.query_params.get("error"),
}
)
@app.post("/settings")
async def save_settings(
request: Request,
odoo_url: str = Form(...),
odoo_db: str = Form(...),
odoo_username: str = Form(...),
odoo_api_key: str = Form(...),
):
"""Save settings to .env file."""
config = {
"ODOO_URL": odoo_url.strip().rstrip('/'),
"ODOO_DB": odoo_db.strip(),
"ODOO_USERNAME": odoo_username.strip(),
"ODOO_API_KEY": odoo_api_key.strip(),
}
try:
save_odoo_config(config)
return RedirectResponse(url="/settings?saved=1", status_code=303)
except Exception as e:
return RedirectResponse(url=f"/settings?error={str(e)}", status_code=303)
@app.get("/test-connection")
async def test_connection():
"""Test Odoo connection with current settings."""
try:
import server
get_odoo_client()
result = server.test_connection()
return JSONResponse({
"success": True,
"message": markdown_to_html(result)
})
except Exception as e:
return JSONResponse({
"success": False,
"message": f"Connection failed: {str(e)}"
})
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8080)
