89 lines
2.8 KiB
Python
89 lines
2.8 KiB
Python
import os
|
|
import xmlrpc.client
|
|
from mcp.server.fastmcp import FastMCP
|
|
from typing import Dict, Any, List, Optional
|
|
|
|
# Odoo Credentials defaults to Hazard environment
|
|
URL = os.getenv("ODOO_URL", "http://localhost:16001")
|
|
DB = os.getenv("ODOO_DB", "hazard_new")
|
|
USER = os.getenv("ODOO_USER", "admin")
|
|
PASSWORD = os.getenv("ODOO_PASSWORD", "admin")
|
|
|
|
# FastMCP Initialization
|
|
mcp = FastMCP("odoo-mcp-hazard")
|
|
|
|
def _get_uid():
|
|
common = xmlrpc.client.ServerProxy(f'{URL}/xmlrpc/2/common')
|
|
return common.authenticate(DB, USER, PASSWORD, {})
|
|
|
|
def _get_models():
|
|
return xmlrpc.client.ServerProxy(f'{URL}/xmlrpc/2/object')
|
|
|
|
@mcp.tool()
|
|
def odoo_search_read(model: str, domain: List[Any], fields: Optional[List[str]] = None, limit: Optional[int] = None) -> List[Dict[str, Any]]:
|
|
"""
|
|
Search and read records from Odoo.
|
|
Args:
|
|
model: Odoo model name (e.g., 'sale.order', 'product.product')
|
|
domain: Search domain list of tuples (e.g., [['name', '=', '123']])
|
|
fields: Optional list of field names to return
|
|
limit: Optional maximum number of records to return
|
|
"""
|
|
uid = _get_uid()
|
|
if not uid:
|
|
raise Exception("Authentication failed with Odoo")
|
|
|
|
models = _get_models()
|
|
kwargs = {}
|
|
if fields:
|
|
kwargs['fields'] = fields
|
|
if limit:
|
|
kwargs['limit'] = limit
|
|
|
|
return models.execute_kw(DB, uid, PASSWORD, model, 'search_read', [domain], kwargs)
|
|
|
|
@mcp.tool()
|
|
def odoo_create(model: str, vals: Dict[str, Any]) -> int:
|
|
"""
|
|
Create a new record in Odoo.
|
|
Args:
|
|
model: Odoo model name (e.g., 'res.partner')
|
|
vals: Dictionary of field values for the new record
|
|
Returns:
|
|
The ID of the newly created record
|
|
"""
|
|
uid = _get_uid()
|
|
models = _get_models()
|
|
return models.execute_kw(DB, uid, PASSWORD, model, 'create', [vals])
|
|
|
|
@mcp.tool()
|
|
def odoo_write(model: str, ids: List[int], vals: Dict[str, Any]) -> bool:
|
|
"""
|
|
Update existing records in Odoo.
|
|
Args:
|
|
model: Odoo model name
|
|
ids: List of record IDs to update
|
|
vals: Dictionary of field values to update
|
|
"""
|
|
uid = _get_uid()
|
|
models = _get_models()
|
|
return models.execute_kw(DB, uid, PASSWORD, model, 'write', [ids, vals])
|
|
|
|
@mcp.tool()
|
|
def odoo_execute_method(model: str, method: str, ids: List[int], kwargs: Optional[Dict[str, Any]] = None) -> Any:
|
|
"""
|
|
Execute an arbitrary method on an Odoo model.
|
|
Args:
|
|
model: Odoo model name
|
|
method: Method name to execute (e.g., 'action_confirm')
|
|
ids: List of record IDs to apply the method to
|
|
kwargs: Optional keyword arguments for the method
|
|
"""
|
|
uid = _get_uid()
|
|
models = _get_models()
|
|
kwargs = kwargs or {}
|
|
return models.execute_kw(DB, uid, PASSWORD, model, method, [ids], kwargs)
|
|
|
|
if __name__ == "__main__":
|
|
mcp.run()
|