feat: add try-except error handling in MCP tools
This commit is contained in:
@@ -4,7 +4,7 @@ MCP Server for Naliia Module
|
|||||||
|
|
||||||
from contextlib import asynccontextmanager
|
from contextlib import asynccontextmanager
|
||||||
from fastmcp import FastMCP
|
from fastmcp import FastMCP
|
||||||
from typing import List, AsyncIterator
|
from typing import List, AsyncIterator, Any, Dict
|
||||||
import datetime
|
import datetime
|
||||||
|
|
||||||
from tryton_mcp.config import settings
|
from tryton_mcp.config import settings
|
||||||
@@ -20,6 +20,15 @@ async def server_lifespan(server: FastMCP) -> AsyncIterator[None]:
|
|||||||
mcp = FastMCP("Tryton MCP Server", lifespan=server_lifespan)
|
mcp = FastMCP("Tryton MCP Server", lifespan=server_lifespan)
|
||||||
|
|
||||||
|
|
||||||
|
def tryton_call(name: str, args: list) -> Dict[str, Any]:
|
||||||
|
try:
|
||||||
|
client = settings.get_client()
|
||||||
|
response = client.call(name, args)
|
||||||
|
return {"success": True, "data": response}
|
||||||
|
except Exception as e:
|
||||||
|
return {"success": False, "error": str(e)}
|
||||||
|
|
||||||
|
|
||||||
@mcp.tool()
|
@mcp.tool()
|
||||||
def create_schedule(
|
def create_schedule(
|
||||||
professional: int, description: str, customer: int, date: str, service_center: int
|
professional: int, description: str, customer: int, date: str, service_center: int
|
||||||
@@ -35,38 +44,42 @@ def create_schedule(
|
|||||||
}
|
}
|
||||||
]
|
]
|
||||||
|
|
||||||
client = settings.get_client()
|
result = tryton_call("model.naliia.schedule.create", [example_schedule, {}])
|
||||||
name = "model.naliia.schedule.create"
|
|
||||||
args = [example_schedule, {}]
|
|
||||||
response = client.call(name, args)
|
|
||||||
|
|
||||||
return response
|
if not result["success"]:
|
||||||
|
raise Exception(f"Failed to create schedule: {result['error']}")
|
||||||
|
|
||||||
|
return result["data"]
|
||||||
|
|
||||||
|
|
||||||
@mcp.tool()
|
@mcp.tool()
|
||||||
def find_service_centers():
|
def find_service_centers():
|
||||||
client = settings.get_client()
|
result = tryton_call(
|
||||||
|
"model.naliia.service_center.search_read",
|
||||||
|
[[[]], 0, None, None, ["name", "address.street"], {}],
|
||||||
|
)
|
||||||
|
|
||||||
name = "model.naliia.service_center.search_read"
|
if not result["success"]:
|
||||||
args = [[[]], 0, None, None, ["name", "address.street"], {}]
|
raise Exception(f"Failed to find service centers: {result['error']}")
|
||||||
response = client.call(name, args)
|
|
||||||
|
|
||||||
return response
|
return result["data"]
|
||||||
|
|
||||||
|
|
||||||
@mcp.tool()
|
@mcp.tool()
|
||||||
def find_products_and_services():
|
def find_products_and_services():
|
||||||
client = settings.get_client()
|
result = tryton_call(
|
||||||
|
"model.product.product.search_read",
|
||||||
|
[
|
||||||
|
[["active", "=", True], ["salable", "=", True]],
|
||||||
|
0,
|
||||||
|
None,
|
||||||
|
None,
|
||||||
|
["name", "list_price", "description"],
|
||||||
|
{"company": 1},
|
||||||
|
],
|
||||||
|
)
|
||||||
|
|
||||||
name = "model.product.product.search_read"
|
if not result["success"]:
|
||||||
args = [
|
raise Exception(f"Failed to find products: {result['error']}")
|
||||||
[["active", "=", True], ["salable", "=", True]],
|
|
||||||
0,
|
|
||||||
None,
|
|
||||||
None,
|
|
||||||
["name", "list_price", "description"],
|
|
||||||
{"company": 1},
|
|
||||||
]
|
|
||||||
response = client.call(name, args)
|
|
||||||
|
|
||||||
return response
|
return result["data"]
|
||||||
|
|||||||
Reference in New Issue
Block a user