stores.py•4.07 kB
"""Store-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 store discovery tools on the server."""
@server.tool("stores.list", description="Locate store locations for a base site.")
async def stores_list(
ctx: Context,
base_site_id: Annotated[
str,
Field(description="OCC base site identifier.", min_length=1, examples=["powertools-spa"]),
],
query: Annotated[
str | None,
Field(description="Free text location search (e.g. city, country)."),
] = None,
latitude: Annotated[
float | None,
Field(description="Latitude used for geo-search."),
] = None,
longitude: Annotated[
float | None,
Field(description="Longitude used for geo-search."),
] = None,
radius: Annotated[
float | None,
Field(description="Radius in meters used to filter stores."),
] = None,
accuracy: Annotated[
float | None,
Field(description="Accuracy in meters for geo-search results."),
] = None,
current_page: Annotated[
int,
Field(description="Zero-based page index.", ge=0, examples=[0]),
] = 0,
page_size: Annotated[
int,
Field(description="Stores per page (1-100).", ge=1, le=100, examples=[20]),
] = 20,
sort: Annotated[
str | None,
Field(description="Optional store sorting method (e.g. 'asc')."),
] = None,
fields: Annotated[
str | None,
Field(description="Override the fields query parameter (defaults to OCC_DEFAULT_FIELDS)."),
] = None,
) -> JsonObject:
await ctx.report_progress(0, total=2, message="Fetching stores")
try:
payload = await occ_client.get_stores(
base_site_id=base_site_id,
query=query,
latitude=latitude,
longitude=longitude,
radius=radius,
accuracy=accuracy,
current_page=current_page,
page_size=page_size,
sort=sort,
fields=fields,
)
except OccApiError as error:
await handle_occ_error(ctx, "stores.list", error)
await ctx.report_progress(1, total=2, message="Processing store response")
await ctx.report_progress(2, total=2, message="Retrieved stores")
return payload
@server.tool("stores.get", description="Fetch the details of a specific base store.")
async def stores_get(
ctx: Context,
base_site_id: Annotated[
str,
Field(description="OCC base site identifier.", min_length=1, examples=["powertools-spa"]),
],
base_store_uid: Annotated[
str,
Field(description="Base store UID to retrieve.", min_length=1, examples=["powertools"]),
],
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 store '{base_store_uid}' for base site '{base_site_id}'",
)
try:
payload = await occ_client.get_store(
base_site_id=base_site_id,
base_store_uid=base_store_uid,
fields=fields,
)
except OccApiError as error:
await handle_occ_error(ctx, "stores.get", error)
await ctx.report_progress(1, total=1, message="Retrieved store details")
return payload