aserrador
b520d47ab3
- Add Dockerfile using uv for dependency management - Add docker-compose.yaml for container orchestration - Add .dockerignore to optimize builds - Update taskipy start task with http transport and port 3001 - Update README.md and AGENTS.md with Docker documentation
161 lines
4.5 KiB
Python
161 lines
4.5 KiB
Python
"""
|
|
MCP Server for Naliia Module
|
|
"""
|
|
|
|
import logging
|
|
from contextlib import asynccontextmanager
|
|
from datetime import datetime
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from fastmcp import FastMCP
|
|
from pydantic import BaseModel, Field, field_validator
|
|
|
|
from tryton_mcp.config import settings
|
|
from tryton_mcp.services.provider import ServiceProvider
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@asynccontextmanager
|
|
async def server_lifespan(server: FastMCP):
|
|
settings.connect()
|
|
yield
|
|
settings.disconnect()
|
|
|
|
|
|
mcp = FastMCP("Tryton MCP Server", lifespan=server_lifespan)
|
|
|
|
_service_provider: Optional[ServiceProvider] = None
|
|
|
|
|
|
def get_service_provider() -> ServiceProvider:
|
|
global _service_provider
|
|
if _service_provider is None:
|
|
_service_provider = ServiceProvider(settings.get_client())
|
|
return _service_provider
|
|
|
|
|
|
class FindCustomerInput(BaseModel):
|
|
identifier: str = Field(min_length=1, description="Customer identifier code")
|
|
|
|
|
|
class CreateCustomerInput(BaseModel):
|
|
name: str = Field(min_length=1, description="Customer name")
|
|
identifiers: List[tuple[str, List[Dict[str, Any]]]] = Field(
|
|
default_factory=list, description="List of identifier tuples"
|
|
)
|
|
|
|
|
|
class CreateScheduleInput(BaseModel):
|
|
professional: int = Field(gt=0, description="Professional ID")
|
|
description: str = Field(
|
|
min_length=1, max_length=500, description="Appointment description"
|
|
)
|
|
customer: int = Field(gt=0, description="Customer ID")
|
|
date: str = Field(description="Date in ISO format (e.g., 2026-03-15T10:00:00)")
|
|
service_center: int = Field(gt=0, description="Service center ID")
|
|
|
|
@field_validator("date")
|
|
@classmethod
|
|
def validate_date(cls, v: str) -> str:
|
|
try:
|
|
datetime.fromisoformat(v)
|
|
except ValueError:
|
|
raise ValueError(
|
|
f"Invalid date format: '{v}'. Expected ISO format (e.g., '2026-03-15T10:00:00' or '2026-03-15T10:00:00+00:00')"
|
|
)
|
|
return v
|
|
|
|
|
|
class FindServiceCentersInput(BaseModel):
|
|
offset: int = Field(default=0, ge=0, description="Offset for pagination")
|
|
limit: Optional[int] = Field(default=None, ge=1, description="Limit for pagination")
|
|
|
|
|
|
class FindProductsInput(BaseModel):
|
|
offset: int = Field(default=0, ge=0, description="Offset for pagination")
|
|
limit: Optional[int] = Field(default=None, ge=1, description="Limit for pagination")
|
|
|
|
|
|
@mcp.tool()
|
|
def find_customer_by_identifier(identifier: str) -> List[Dict]:
|
|
FindCustomerInput(identifier=identifier)
|
|
|
|
services = get_service_provider()
|
|
result = services.party.find_by_identifier(identifier)
|
|
|
|
if result.is_error:
|
|
raise Exception(f"Customer with identifier {identifier} not found.")
|
|
|
|
return result.data
|
|
|
|
|
|
@mcp.tool()
|
|
def create_customer(
|
|
name: str, identifiers: List[tuple[str, List[Dict[str, Any]]]]
|
|
) -> List[int]:
|
|
CreateCustomerInput(name=name, identifiers=identifiers)
|
|
|
|
services = get_service_provider()
|
|
result = services.party.create(name, identifiers)
|
|
|
|
if result.is_error:
|
|
raise Exception(f"Failed to create {name} as customer: {result.error}")
|
|
|
|
return result.data
|
|
|
|
|
|
@mcp.tool()
|
|
def create_schedule(
|
|
professional: int, description: str, customer: int, date: str, service_center: int
|
|
) -> List[int]:
|
|
CreateScheduleInput(
|
|
professional=professional,
|
|
description=description,
|
|
customer=customer,
|
|
date=date,
|
|
service_center=service_center,
|
|
)
|
|
|
|
services = get_service_provider()
|
|
result = services.schedule.create(
|
|
professional, description, customer, date, service_center
|
|
)
|
|
|
|
if result.is_error:
|
|
raise Exception(f"Failed to create schedule: {result.error}")
|
|
|
|
return result.data
|
|
|
|
|
|
@mcp.tool()
|
|
def find_service_centers(offset: int = 0, limit: Optional[int] = None) -> List[Dict]:
|
|
FindServiceCentersInput(offset=offset, limit=limit)
|
|
|
|
services = get_service_provider()
|
|
result = services.service_center.find_all(offset=offset, limit=limit)
|
|
|
|
if result.is_error:
|
|
raise Exception(f"Failed to find service centers: {result.error}")
|
|
|
|
return result.data
|
|
|
|
|
|
@mcp.tool()
|
|
def find_products_and_services(
|
|
offset: int = 0, limit: Optional[int] = None
|
|
) -> List[Dict]:
|
|
FindProductsInput(offset=offset, limit=limit)
|
|
|
|
services = get_service_provider()
|
|
result = services.product.find_salable(offset=offset, limit=limit)
|
|
|
|
if result.is_error:
|
|
raise Exception(f"Failed to find products: {result.error}")
|
|
|
|
return result.data
|