categories.py•3.39 kB
"""Category-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 category tools on the server."""
@server.tool("categories.products", description="Fetch products for a category within a base site.")
async def categories_products(
ctx: Context,
base_site_id: Annotated[
str,
Field(description="The OCC base site identifier.", min_length=1, examples=["electronics"]),
],
category_id: Annotated[
str,
Field(description="Category identifier to browse.", min_length=1, examples=["cameras"]),
],
current_page: Annotated[
int,
Field(
description="Zero-based page index.",
ge=0,
examples=[0],
),
] = 0,
page_size: Annotated[
int,
Field(
description="Number of products per page (1-100).",
ge=1,
le=100,
examples=[20],
),
] = 20,
sort: Annotated[
str | None,
Field(
description="Sort expression supported by OCC (e.g. 'name:asc').",
examples=["name:asc"],
),
] = None,
query: Annotated[
str | None,
Field(
description="Optional free-text filter applied within the category.",
examples=["camera"],
),
] = None,
fields: Annotated[
str | None,
Field(
description="Override the fields query parameter (defaults to OCC_DEFAULT_FIELDS).",
examples=["DEFAULT"],
),
] = None,
) -> JsonObject:
await ctx.report_progress(
0,
total=3,
message=(
f"Fetching products for category '{category_id}' in base site '{base_site_id}' "
f"(page {current_page}, size {page_size})"
),
)
params = {
"base_site_id": base_site_id,
"category_id": category_id,
"current_page": current_page,
"page_size": page_size,
"sort": sort,
"query": query,
"fields": fields,
}
await ctx.report_progress(1, total=3, message="Submitting request to OCC")
try:
payload = await occ_client.get_products_by_category(
base_site_id=base_site_id,
category_id=category_id,
current_page=current_page,
page_size=page_size,
sort=sort,
query=query,
fields=fields,
)
except OccApiError as error:
await handle_occ_error(ctx, "categories.products", error)
await ctx.report_progress(2, total=3, message="Processing OCC response")
await ctx.report_progress(3, total=3, message="Retrieved category products")
await ctx.info("categories.products completed", extra=params)
return payload