chore: clean up historical scripts, add MCP configurations, update agent rules
This commit is contained in:
@@ -0,0 +1 @@
|
||||
mcp
|
||||
@@ -0,0 +1,88 @@
|
||||
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()
|
||||
Reference in New Issue
Block a user