mcp

app.py

@@ -1,7 +1,452 @@
+import xmlrpc.client
 import gradio as gr
+import os
+import json # For parsing args/kwargs in execute_kw
+from dotenv import load_dotenv
 
-def greet(name):
-    return "Hello " + name + "!!"
+# Load environment variables from .env file
+load_dotenv()
 
-demo = gr.Interface(fn=greet, inputs="text", outputs="text")
-demo.launch()
+# --- Odoo XML-RPC Helper Functions ---
+def authenticate(url, db, username, password):
+    """
+    Authenticate with Odoo using XML-RPC.
+    Returns user ID (int) or False.
+    """
+    try:
+        common = xmlrpc.client.ServerProxy(f'{url}/xmlrpc/2/common')
+        uid = common.authenticate(db, username, password, {})
+        return uid if uid else False
+    except Exception as e:
+        print(f"Authentication error: {e}")
+        return False
+
+def list_models(url, db, uid, password):
+    """
+    Returns list of (display_name (technical_name), technical_name) tuples.
+    """
+    try:
+        models_proxy = xmlrpc.client.ServerProxy(f'{url}/xmlrpc/2/object')
+        model_data = models_proxy.execute_kw(db, uid, password, 'ir.model', 'search_read', 
+                                             [[]], {'fields': ['model', 'name'], 'order': 'name'})
+        return sorted([(f"{m['name']} ({m['model']})", m['model']) for m in model_data if m.get('model') and m.get('name')], key=lambda x: x[0])
+    except Exception as e:
+        print(f"List models error: {e}")
+        return []
+
+def get_model_fields(url, db, uid, password, model_name):
+    """
+    Returns dictionary of fields for a model.
+    """
+    try:
+        models_proxy = xmlrpc.client.ServerProxy(f'{url}/xmlrpc/2/object')
+        fields = models_proxy.execute_kw(db, uid, password, model_name, 'fields_get', [], 
+                                         {'attributes': ['string', 'type', 'help', 'readonly', 'required', 'relation', 'selection']})
+        return fields
+    except Exception as e:
+        print(f"Get model fields error for {model_name}: {e}")
+        return {"error": f"Failed to fetch fields for {model_name}: {str(e)}"}
+
+def search_read(url, db, uid, password, model_name, domain, limit):
+    """
+    Returns list of records matching domain and limit.
+    """
+    try:
+        models_proxy = xmlrpc.client.ServerProxy(f'{url}/xmlrpc/2/object')
+        records = models_proxy.execute_kw(db, uid, password, model_name, 'search_read', 
+                                         [domain], {'limit': limit})  # Removed 'raise_exception'
+        return records
+    except Exception as e:
+        print(f"Search read error for {model_name}: {e}")
+        return {"error": f"Failed to search_read {model_name}: {str(e)}"}
+
+def search(url, db, uid, password, model_name, domain):
+    """
+    Returns list of record IDs matching domain.
+    """
+    try:
+        models_proxy = xmlrpc.client.ServerProxy(f'{url}/xmlrpc/2/object')
+        record_ids = models_proxy.execute_kw(db, uid, password, model_name, 'search', 
+                                             [domain])  # Removed 'raise_exception'
+        return record_ids
+    except Exception as e:
+        print(f"Search error for {model_name}: {e}")
+        return {"error": f"Failed to search {model_name}: {str(e)}"}
+
+def read(url, db, uid, password, model_name, record_id):
+    """
+    Returns complete record details for a given ID.
+    """
+    try:
+        models_proxy = xmlrpc.client.ServerProxy(f'{url}/xmlrpc/2/object')
+        # Odoo's read method expects a list of IDs, even if it's just one.
+        record = models_proxy.execute_kw(db, uid, password, model_name, 'read', [[record_id]])  # Removed 'raise_exception'
+        return record[0] if record else {"error": f"Record ID {record_id} not found or access denied."}
+    except Exception as e:
+        print(f"Read error for {model_name} ID {record_id}: {e}")
+        return {"error": f"Failed to read {model_name} ID {record_id}: {str(e)}"}
+
+def execute_kw(url, db, uid, password, model_name, method, args, kwargs):
+    """
+    Execute a generic method on a model.
+    Filters out 'raise_exception' for methods that do not support it (e.g., create, write).
+    """
+    try:
+        models_proxy = xmlrpc.client.ServerProxy(f'{url}/xmlrpc/2/object')
+        # Only allow 'raise_exception' for methods that support it
+        methods_with_raise_exception = ["search", "search_read", "read"]
+        if method not in methods_with_raise_exception:
+            # Remove 'raise_exception' if present
+            kwargs.pop('raise_exception', None)
+        result = models_proxy.execute_kw(db, uid, password, model_name, method, args, kwargs)
+        return result
+    except Exception as e:
+        print(f"Execution error for {model_name}.{method}: {e}")
+        return {"error": f"Failed to execute {method} on {model_name}: {str(e)}"}
+
+# --- Gradio Application with gr.Blocks ---
+with gr.Blocks(title="Odoo Inspector", theme=gr.themes.Soft()) as demo:
+    # State variables
+    state_uid = gr.State(None)
+    state_odoo_url = gr.State(None)
+    state_odoo_db = gr.State(None)
+    state_odoo_password = gr.State(None)
+
+    gr.Markdown("## 1. Odoo Connection Details")
+    with gr.Row():
+        input_odoo_url = gr.Textbox(label="Odoo URL", value=os.getenv("ODOO_URL", ""), interactive=True)
+        input_odoo_db = gr.Textbox(label="Database Name", value=os.getenv("ODOO_DB", ""), interactive=True)
+    with gr.Row():
+        input_odoo_username = gr.Textbox(label="Username", value=os.getenv("ODOO_USERNAME", ""), interactive=True)
+        input_odoo_password = gr.Textbox(label="Password", value=os.getenv("ODOO_PASSWORD", ""), interactive=True)
+    connect_button = gr.Button("Connect and Fetch Models")
+    connection_status_output = gr.Textbox(label="Status", interactive=False)
+
+    with gr.Group(visible=False) as model_inspector_group:
+        gr.Markdown("## 2. Inspect Model Fields")
+        # The choices for model_dropdown will be dynamically updated by the connection step
+        model_dropdown = gr.Dropdown(label="Select a Model", choices=[], interactive=True, allow_custom_value=True)
+        model_name_api = gr.Textbox(label="Model Name (API)", visible=False)
+        get_fields_button = gr.Button("Get Fields for Selected Model")
+        get_fields_api_button = gr.Button(visible=False)  # Dummy button for MCP binding
+        model_fields_output = gr.JSON(label="Model Fields")
+
+    with gr.Group(visible=False) as record_ops_group:
+        gr.Markdown("## 3. Record Operations")
+        gr.Markdown("""
+        Use this section to find and view records from the selected Odoo model.
+        
+        **Domain**: Specify search criteria as a Python list of lists/tuples. 
+        - Example: `[['is_company', '=', True], ['customer_rank', '>', 0]]` (finds companies that are customers).
+        - Use `[]` to match all records (respecting Odoo's default limits if any).
+        - The input is evaluated as Python code, so ensure correct syntax.
+        
+        **Limit**: Max number of records for 'Search & Read Records'.
+        
+        **Buttons**:
+        - `Search & Read Records`: Fetches full records matching the domain and limit.
+        - `Search Record IDs`: Fetches only the IDs of records matching the domain. Populates the 'Select Record ID' dropdown.
+        - `Read Selected Record Details`: Fetches full details for the ID chosen in 'Select Record ID'.
+        
+        **Select Record ID**: Choose an ID from the list populated by `Search Record IDs`, or type a specific record ID and press Enter/Return before clicking `Read Selected Record Details`.
+        """)
+        domain_input = gr.Textbox(label="Domain (Python list of lists/tuples)", placeholder="[['field_name', '=', 'value']] or [] for all", value="[]")
+        limit_input = gr.Number(label="Limit (for Search & Read)", value=10, minimum=1, step=1)
+        with gr.Row():
+            search_read_button = gr.Button("Search & Read Records")
+            search_button = gr.Button("Search Record IDs")
+        record_id_dropdown = gr.Dropdown(label="Select Record ID (from Search results or type ID)", choices=[], interactive=True, allow_custom_value=True)
+        read_button = gr.Button("Read Selected Record Details")
+        record_operation_output = gr.JSON(label="Record Operation Result")
+
+    with gr.Group(visible=False) as exec_kw_group:
+        gr.Markdown("## 4. Execute Generic Odoo Method")
+        gr.Markdown("""
+        This section allows you to call almost any method on the selected Odoo model via `execute_kw`.
+        
+        **Method Name**: The technical name of the method to call (e.g., `check_access_rights`, `name_search`, `default_get`).
+        
+        **Arguments (JSON list)**: Positional arguments for the method, provided as a JSON-formatted list.
+        - Example for `check_access_rights`: `["read", false]`
+        - Example for `name_search`: `[]` (if using named arguments in Kwargs for criteria)
+        - If no positional arguments, use `[]`.
+        
+        **Keyword Arguments (JSON dict)**: Named arguments for the method, provided as a JSON-formatted dictionary.
+        - Example for `check_access_rights`: `{"raise_exception": false}`
+        - Example for `name_search`: `{"name": "Azure", "operator": "ilike", "limit": 8}`
+        - It's often useful to include `"raise_exception": false` to prevent Odoo from stopping the XML-RPC call on application errors, allowing the app to display the error instead.
+        - For context: `{"context": {"lang": "en_US", "active_test": false}}`
+        - If no keyword arguments, use `{}`.
+        """)
+        method_name_input = gr.Textbox(label="Method Name", placeholder="e.g., check_access_rights")
+        args_input = gr.Textbox(label="Arguments (JSON list)", placeholder='[\"read\", false]', value="[]", lines=2)
+        kwargs_input = gr.Textbox(label="Keyword Arguments (JSON dict)", placeholder='{"raise_exception": false, "context": {"lang": "en_US"}}', value="{}", lines=3)
+        execute_kw_button = gr.Button("Execute Method")
+        execute_kw_output = gr.JSON(label="Method Execution Result")
+
+    # --- Event Handlers ---
+    def handle_connect_and_fetch_models():
+        """
+        Establishes a connection to the Odoo server and fetches available models for inspection using credentials from environment variables.
+        Returns: list: UI updates including connection status and available models.
+        """
+        url = os.environ.get("ODOO_URL")
+        db = os.environ.get("ODOO_DB")
+        username = os.environ.get("ODOO_USERNAME")
+        password = os.environ.get("ODOO_PASSWORD")
+        uid_val = authenticate(url, db, username, password)
+        updates = {
+            connection_status_output: "",
+            model_dropdown: gr.update(choices=[], value=None),
+            model_inspector_group: gr.update(visible=False),
+            record_ops_group: gr.update(visible=False),
+            exec_kw_group: gr.update(visible=False)
+        }
+        if not uid_val:
+            updates[connection_status_output] = "Authentication failed. Check credentials and Odoo connection."
+            return list(updates.values())
+
+        models_choices = list_models(url, db, uid_val, password)
+        if not models_choices:
+            updates[connection_status_output] = f"Authenticated (UID: {uid_val}), but failed to fetch models or no models found."
+            return list(updates.values())
+
+        updates[connection_status_output] = f"Successfully connected. UID: {uid_val}. Models loaded."
+        updates[model_dropdown] = gr.update(choices=models_choices, value=models_choices[0][1] if models_choices else None, interactive=True)
+        updates[model_inspector_group] = gr.update(visible=True)
+        updates[record_ops_group] = gr.update(visible=True)
+        updates[exec_kw_group] = gr.update(visible=True)
+        return list(updates.values())
+
+    connect_button.click(
+        fn=handle_connect_and_fetch_models,
+        inputs=[],
+        outputs=[connection_status_output, model_dropdown, model_inspector_group, record_ops_group, exec_kw_group],
+        api_name="connect_odoo"
+    )
+
+    def handle_get_model_fields_gui(selected_model):
+        """
+        GUI handler: Retrieves the fields and their metadata for the selected Odoo model using credentials from environment variables.
+        Uses dropdown for model selection.
+        """
+        url = os.environ.get("ODOO_URL")
+        db = os.environ.get("ODOO_DB")
+        username = os.environ.get("ODOO_USERNAME")
+        password = os.environ.get("ODOO_PASSWORD")
+        uid = authenticate(url, db, username, password)
+        if not all([selected_model, uid, url, db, password]):
+            return {"error": "Connection/model state missing. Reconnect."}
+        return get_model_fields(url, db, uid, password, selected_model)
+
+    def handle_get_model_fields_api(selected_model):
+        """
+        MCP/API handler: Accepts any string for selected_model (from textbox, not dropdown).
+        """
+        url = os.environ.get("ODOO_URL")
+        db = os.environ.get("ODOO_DB")
+        username = os.environ.get("ODOO_USERNAME")
+        password = os.environ.get("ODOO_PASSWORD")
+        uid = authenticate(url, db, username, password)
+        if not all([selected_model, uid, url, db, password]):
+            return {"error": "Connection/model state missing. Reconnect."}
+        return get_model_fields(url, db, uid, password, selected_model)
+
+    # GUI binding (dropdown)
+    get_fields_button.click(
+        fn=handle_get_model_fields_gui,
+        inputs=[model_dropdown],
+        outputs=[model_fields_output],
+    )
+    # MCP/API binding (textbox, hidden in GUI)
+    get_fields_api_button.click(
+        fn=handle_get_model_fields_api,
+        inputs=[model_name_api],
+        outputs=[model_fields_output],
+        api_name="get_model_fields"
+    )
+
+    def handle_search_read_click(selected_model, domain_str, limit):
+        """
+        Searches for records in the selected model matching the provided domain and returns their details (search_read) using credentials from environment variables.
+        """
+        url = os.environ.get("ODOO_URL")
+        db = os.environ.get("ODOO_DB")
+        username = os.environ.get("ODOO_USERNAME")
+        password = os.environ.get("ODOO_PASSWORD")
+        uid = authenticate(url, db, username, password)
+        if not all([selected_model, uid, url, db, password]):
+            return {"error": "Connection/model state missing."}
+        try:
+            domain = eval(domain_str) # Caution: eval can be risky
+        except Exception as e:
+            return {"error": f"Invalid domain: {e}"}
+        return search_read(url, db, uid, password, selected_model, domain, int(limit))
+
+    search_read_button.click(
+        fn=handle_search_read_click,
+        inputs=[model_dropdown, domain_input, limit_input],
+        outputs=[record_operation_output],
+        api_name="search_read"
+    )
+
+    def handle_search_click(selected_model: str, domain_str: str):
+        """
+        Searches for record IDs in the selected Odoo model matching the provided domain using credentials from environment variables.
+        Args:
+            selected_model (str): The technical name of the Odoo model to search (e.g., 'res.partner').
+            domain_str (str): The search domain as a Python list-of-lists string (e.g., "[['is_company', '=', True]]").
+        Returns:
+            tuple: (result_dict, gradio_update)
+                - result_dict (dict): Contains either the number of record IDs found or an error message.
+                - gradio_update: Gradio Dropdown update object for record ID selection.
+        """
+        url = os.environ.get("ODOO_URL")
+        db = os.environ.get("ODOO_DB")
+        username = os.environ.get("ODOO_USERNAME")
+        password = os.environ.get("ODOO_PASSWORD")
+        uid = authenticate(url, db, username, password)
+        if not all([selected_model, uid, url, db, password]):
+            return {"error": "Connection/model state missing."}, gr.update(choices=[])
+        try:
+            domain = eval(domain_str)
+        except Exception as e:
+            return {"error": f"Invalid domain: {e}"}, gr.update(choices=[])
+        record_ids = search(url, db, uid, password, selected_model, domain)
+        if isinstance(record_ids, dict) and 'error' in record_ids:
+            return record_ids, gr.update(choices=[])
+        if isinstance(record_ids, list):
+            id_choices = [(str(rid), rid) for rid in record_ids]
+            return {"record_ids_found": len(record_ids)}, gr.update(choices=id_choices, value=id_choices[0][1] if id_choices else None)
+        return {"error": "Unexpected search result"}, gr.update(choices=[])
+
+    search_button.click(
+        fn=handle_search_click,
+        inputs=[model_dropdown, domain_input],
+        outputs=[record_operation_output, record_id_dropdown],
+        api_name="search_ids"
+    )
+
+    def handle_read_click(selected_model: str, record_id: str):
+        """
+        Reads and returns the details for a specific record ID from the selected Odoo model using credentials from environment variables.
+        Args:
+            selected_model (str): The technical name of the Odoo model to read from (e.g., 'res.partner').
+            record_id (str): The ID of the record to read (should be convertible to int).
+        Returns:
+            dict: Record details as returned by Odoo, or an error message dict.
+        """
+        url = os.environ.get("ODOO_URL")
+        db = os.environ.get("ODOO_DB")
+        username = os.environ.get("ODOO_USERNAME")
+        password = os.environ.get("ODOO_PASSWORD")
+        uid = authenticate(url, db, username, password)
+        if not all([selected_model, record_id, uid, url, db, password]):
+            return {"error": "Connection/model/ID state missing."}
+        try:
+            rec_id = int(record_id)
+        except ValueError:
+            return {"error": f"Invalid Record ID: {record_id}. Must be int."}
+        return read(url, db, uid, password, selected_model, rec_id)
+
+    read_button.click(
+        fn=handle_read_click,
+        inputs=[model_dropdown, record_id_dropdown],
+        outputs=[record_operation_output],
+        api_name="read_record"
+    )
+
+    def handle_execute_kw_click(selected_model: str, method_name: str, args_str: str = "[]", kwargs_str: str = "{}"):
+        """
+        Executes a generic method on the selected Odoo model using the provided method name, positional arguments, and keyword arguments.
+        Credentials are taken from environment variables.
+        Args:
+            selected_model (str): The technical name of the Odoo model (e.g., 'res.partner').
+            method_name (str): The technical name of the method to call (e.g., 'search', 'create').
+            args_str (str, optional): JSON string representing a list of positional arguments. Defaults to '[]'.
+            kwargs_str (str, optional): JSON string representing a dictionary of keyword arguments. Defaults to '{}'.
+        Returns:
+            dict: The result of the method call as returned by Odoo, or an error message dict.
+        """
+        url = os.environ.get("ODOO_URL")
+        db = os.environ.get("ODOO_DB")
+        username = os.environ.get("ODOO_USERNAME")
+        password = os.environ.get("ODOO_PASSWORD")
+        uid = authenticate(url, db, username, password)
+        if not all([selected_model, method_name, uid, url, db, password]):
+            return {"error": "Connection/model/method state missing."}
+        try:
+            args = json.loads(args_str)
+        except Exception as e:
+            return {"error": f"Invalid JSON in Args: {e}"}
+        try:
+            kwargs = json.loads(kwargs_str)
+        except Exception as e:
+            return {"error": f"Invalid JSON in Kwargs: {e}"}
+        return execute_kw(url, db, uid, password, selected_model, method_name, args, kwargs)
+
+    execute_kw_button.click(
+        fn=handle_execute_kw_click,
+        inputs=[model_dropdown, method_name_input, args_input, kwargs_input],
+        outputs=[execute_kw_output],
+        api_name="execute_kw"
+    )
+
+    # --- New Order Creation Section ---
+    with gr.Group(visible=True) as order_creation_group:
+        gr.Markdown("## 5. Create New Order (Generic)")
+        gr.Markdown("""
+        Use this section to create a new order by specifying order lines and a note. Order lines should be provided as a JSON list of dicts, e.g.:
+        
+        ```json
+        [
+          {"product_id": 1, "product_uom_qty": 2, "price_unit": 10.0},
+          {"product_id": 2, "product_uom_qty": 1, "price_unit": 5.5}
+        ]
+        ```
+        Note is a free-form string. All fields are passed to Odoo's `sale.order` model via `create`.
+        """)
+        order_lines_input = gr.Textbox(label="Order Lines (JSON list of dicts)", placeholder='[{"product_id": 1, "product_uom_qty": 2, "price_unit": 10.0}]', lines=4)
+        note_input = gr.Textbox(label="Order Note", placeholder="Enter order note", lines=2)
+        create_order_button = gr.Button("Create Order")
+        create_order_output = gr.JSON(label="Order Creation Result")
+
+    def create_order_generic(order_lines_json: str, note: str):
+        """
+        Create a new order in Odoo using the sale.order model and the create method, with credentials from environment variables.
+        Args:
+            order_lines_json (str): JSON string representing a list of order line dicts. Example:
+                '[{"product_id": 1, "product_uom_qty": 2, "price_unit": 10.0}]'
+            note (str): Optional note to attach to the order.
+        Returns:
+            dict: Contains the created order ID (as 'order_id') if successful, or an error message dict.
+        """
+        url = os.environ.get("ODOO_URL")
+        db = os.environ.get("ODOO_DB")
+        username = os.environ.get("ODOO_USERNAME")
+        password = os.environ.get("ODOO_PASSWORD")
+        uid = authenticate(url, db, username, password)
+        try:
+            order_lines = json.loads(order_lines_json)
+            if not isinstance(order_lines, list):
+                return {"error": "Order lines must be a JSON list of dicts."}
+        except Exception as e:
+            return {"error": f"Invalid JSON in order lines: {e}"}
+        order_data = {
+            "order_line": [
+                (0, 0, line) for line in order_lines
+            ],
+            "note": note or ""
+        }
+        result = execute_kw(url, db, uid, password, "sale.order", "create", [order_data], {"context": {"lang": "en_US"}})
+        return {"order_id": result} if isinstance(result, int) else result
+
+    create_order_button.click(
+        fn=create_order_generic,
+        inputs=[order_lines_input, note_input],
+        outputs=[create_order_output],
+        api_name="create_order"
+    )
+
+if __name__ == "__main__":
+    demo.launch(mcp_server=True)

requirements.txt

@@ -0,0 +1,4 @@
+gradio>=5.43.1
+requests>=2.32.5
+pydantic>=2.11.1
+python-dotenv>=1.1.1