products.py•4.67 kB
"""Product-related FastMCP tools."""
from __future__ import annotations
from typing import Annotated
from fastmcp import Context, FastMCP
from pydantic import Field
from app.models import JsonObject
from app.occ_client import occ_client
from app.utils import OccApiError
from .common import handle_occ_error
def register(server: FastMCP) -> None:
"""Register product tools on the FastMCP server."""
@server.tool("products.get", description="Retrieve product details by code.")
async def products_get(
ctx: Context,
base_site_id: Annotated[
str,
Field(description="OCC base site identifier.", min_length=1, examples=["powertools-spa"]),
],
product_code: Annotated[
str,
Field(description="Product code identifier.", min_length=1, examples=["872912"]),
],
fields: Annotated[
str | None,
Field(description="Override the fields query parameter (defaults to OCC_DEFAULT_FIELDS)."),
] = None,
) -> JsonObject:
await ctx.report_progress(0, total=1, message=f"Fetching product '{product_code}'")
try:
payload = await occ_client.get_product(
base_site_id=base_site_id,
product_code=product_code,
fields=fields,
)
except OccApiError as error:
await handle_occ_error(ctx, "products.get", error)
await ctx.report_progress(1, total=1, message="Retrieved product")
return payload
@server.tool("products.stock", description="Retrieve stock information for a product.")
async def products_stock(
ctx: Context,
base_site_id: Annotated[
str,
Field(description="OCC base site identifier.", min_length=1, examples=["powertools-spa"]),
],
product_code: Annotated[
str,
Field(description="Product code identifier.", min_length=1, examples=["872912"]),
],
location: Annotated[
str | None,
Field(description="Optional free-text location filter."),
] = None,
latitude: Annotated[
float | None,
Field(description="Optional latitude coordinate for geo-based stock search."),
] = None,
longitude: Annotated[
float | None,
Field(description="Optional longitude coordinate for geo-based stock search."),
] = None,
) -> JsonObject:
await ctx.report_progress(0, total=1, message="Fetching product stock")
try:
payload = await occ_client.get_product_stock(
base_site_id=base_site_id,
product_code=product_code,
location=location,
latitude=latitude,
longitude=longitude,
)
except OccApiError as error:
await handle_occ_error(ctx, "products.stock", error)
await ctx.report_progress(1, total=1, message="Retrieved product stock")
return payload
@server.tool(
"products.stockCount",
description="Retrieve aggregate stock level metadata for a product (HEAD request).",
)
async def products_stock_count(
ctx: Context,
base_site_id: Annotated[
str,
Field(description="OCC base site identifier.", min_length=1, examples=["powertools-spa"]),
],
product_code: Annotated[
str,
Field(description="Product code identifier.", min_length=1, examples=["872912"]),
],
location: Annotated[
str | None,
Field(description="Optional free-text location filter."),
] = None,
latitude: Annotated[
float | None,
Field(description="Optional latitude coordinate for geo-based stock search."),
] = None,
longitude: Annotated[
float | None,
Field(description="Optional longitude coordinate for geo-based stock search."),
] = None,
) -> JsonObject:
await ctx.report_progress(0, total=1, message="Fetching stock count metadata")
try:
metadata = await occ_client.head_product_stock(
base_site_id=base_site_id,
product_code=product_code,
location=location,
latitude=latitude,
longitude=longitude,
)
except OccApiError as error:
await handle_occ_error(ctx, "products.stockCount", error)
await ctx.report_progress(1, total=1, message="Retrieved stock count metadata")
return metadata