125 lines
3.2 KiB
Python
125 lines
3.2 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import os
|
|
import xmlrpc.client
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
from fastmcp import FastMCP
|
|
|
|
mcp = FastMCP("Odoo MCP Server")
|
|
|
|
def get_odoo_client():
|
|
"""
|
|
Get Odoo XML-RPC client connection and authentication details.
|
|
"""
|
|
|
|
url = os.environ.get("ODOO_URL", "http://localhost:8069")
|
|
db = os.environ.get("ODOO_DATABASE", "odoo")
|
|
username = os.environ.get("ODOO_USERNAME", "admin")
|
|
password = os.environ.get("ODOO_PASSWORD", "admin")
|
|
|
|
common = xmlrpc.client.ServerProxy(f"{url}/xmlrpc/2/common")
|
|
|
|
uid = common.authenticate(db, username, password, {})
|
|
if not uid:
|
|
raise Exception(f"Authentication failed for user '{username}' on database '{db}'")
|
|
|
|
models = xmlrpc.client.ServerProxy(f"{url}/xmlrpc/2/object")
|
|
|
|
return models, db, uid, password
|
|
|
|
def execute_odoo_method(model: str, method: str, *args, **kwargs) -> Any:
|
|
"""
|
|
Execute an Odoo method via XML-RPC.
|
|
"""
|
|
models, db, uid, password = get_odoo_client()
|
|
|
|
models.execute_kw(db, uid, password, model, "check_access_rights",
|
|
["read"], {"raise_exception": False})
|
|
|
|
return models.execute_kw(db, uid, password, model, method, list(args), kwargs)
|
|
|
|
@mcp.tool()
|
|
def create_contact(
|
|
name: str,
|
|
email: str = None,
|
|
phone: str = None,
|
|
mobile: str = None,
|
|
street: str = None,
|
|
city: str = None,
|
|
zip: str = None,
|
|
country_id: int = None,
|
|
is_company: bool = None
|
|
):
|
|
"""
|
|
Create a new contact (res.partner) in Odoo.
|
|
"""
|
|
try:
|
|
contact_data = {"name": name}
|
|
optional_fields = {
|
|
"email": email,
|
|
"phone": phone,
|
|
"mobile": mobile,
|
|
"street": street,
|
|
"city": city,
|
|
"zip": zip,
|
|
"country_id": country_id,
|
|
"is_company": is_company
|
|
}
|
|
|
|
for field, value in optional_fields.items():
|
|
if value is not None:
|
|
contact_data[field] = value
|
|
|
|
contact_id = execute_odoo_method("res.partner", "create", contact_data)
|
|
|
|
return f"Successfully created contact '{name}' with ID {contact_id}"
|
|
|
|
except Exception as e:
|
|
return f"Error creating contact: {str(e)}"
|
|
|
|
@mcp.tool()
|
|
def get_models():
|
|
"""Retrieve a list of available models in Odoo."""
|
|
try:
|
|
models = execute_odoo_method(
|
|
"ir.model",
|
|
"search_read",
|
|
[],
|
|
fields=["name", "model"]
|
|
)
|
|
|
|
return models
|
|
|
|
except Exception as e:
|
|
return [{"error": f"Error retrieving models: {str(e)}"}]
|
|
|
|
@mcp.tool()
|
|
def execute_custom_method(
|
|
model: str,
|
|
method: str,
|
|
args: List = None,
|
|
kwargs: Dict = None
|
|
):
|
|
"""
|
|
Execute a custom Odoo method with arbitrary arguments.
|
|
"""
|
|
try:
|
|
if args is None:
|
|
args = []
|
|
if kwargs is None:
|
|
kwargs = {}
|
|
|
|
result = execute_odoo_method(model, method, *args, **kwargs)
|
|
|
|
# Convert result to JSON-serializable format
|
|
if isinstance(result, (list, dict, str, int, float, bool, type(None))):
|
|
return result
|
|
else:
|
|
return str(result)
|
|
|
|
except Exception as e:
|
|
return {"error": f"Error executing method: {str(e)}"}
|
|
|
|
if __name__ == "__main__":
|
|
mcp.run()
|