diff --git a/statuspro_mcp_server/src/statuspro_mcp/tools/orders.py b/statuspro_mcp_server/src/statuspro_mcp/tools/orders.py index 29cba59..9add33f 100644 --- a/statuspro_mcp_server/src/statuspro_mcp/tools/orders.py +++ b/statuspro_mcp_server/src/statuspro_mcp/tools/orders.py @@ -30,6 +30,7 @@ import asyncio import logging +from collections.abc import Awaitable from typing import Annotated, Any from fastmcp import Context, FastMCP @@ -43,6 +44,7 @@ CoercedStrList, CoercedStrListOpt, ) +from statuspro_mcp.tools.param_types import ConfirmFlag, OrderIdParam from statuspro_mcp.tools.prefab_ui import ( build_bulk_status_change_preview_ui, build_comment_preview_ui, @@ -82,6 +84,7 @@ set_order_due_date, update_order_status as update_order_status_api, ) +from statuspro_public_api_client.domain.converters import to_unset from statuspro_public_api_client.models.add_order_comment_request import ( AddOrderCommentRequest, ) @@ -146,7 +149,7 @@ def _history_entry(item: Any) -> HistoryEntry: async def _bounded_gather[T]( - coros: list[Any], *, limit: int = _BATCH_CONCURRENCY_LIMIT + coros: list[Awaitable[T]], *, limit: int = _BATCH_CONCURRENCY_LIMIT ) -> list[T | Exception]: """Run ``coros`` with bounded concurrency; mirrors ``asyncio.gather`` shape. @@ -158,7 +161,7 @@ async def _bounded_gather[T]( """ sem = asyncio.Semaphore(limit) - async def _run(coro: Any) -> Any: + async def _run(coro: Awaitable[T]) -> T | Exception: async with sem: try: return await coro @@ -561,7 +564,7 @@ async def fetch_for_code(code: str) -> list[Any]: ) async def get_order( context: Context, - order_id: Annotated[int, Field(description="StatusPro order id")], + order_id: OrderIdParam, history_limit: Annotated[ int, Field( @@ -759,7 +762,7 @@ async def count_by_status(code: str | None, name: str | None) -> StatusCount: # endpoint and turn one rate-limit hit into 20 retries in lockstep. sem = asyncio.Semaphore(_BATCH_CONCURRENCY_LIMIT) - async def bounded[T](coro: Any) -> T: + async def bounded[T](coro: Awaitable[T]) -> T: async with sem: return await coro @@ -810,7 +813,7 @@ async def bounded[T](coro: Any) -> T: ) async def get_order_history( context: Context, - order_id: Annotated[int, Field(description="StatusPro order id")], + order_id: OrderIdParam, page: Annotated[ int, Field(description="1-based page number.", ge=1), @@ -849,7 +852,7 @@ async def get_order_history( ) async def update_order_status( context: Context, - order_id: int, + order_id: OrderIdParam, status_code: Annotated[ str, Field(description="8-char status code, e.g. 'st000002'") ], @@ -865,9 +868,7 @@ async def update_order_status( email_additional: Annotated[ bool, Field(description="Send additional notification emails") ] = True, - confirm: Annotated[ - bool, Field(description="Must be true to apply the change") - ] = False, + confirm: ConfirmFlag = False, ) -> ToolResult: services = get_services(context) @@ -927,7 +928,7 @@ async def update_order_status( body = UpdateOrderStatusRequest( status_code=status_code, - comment=comment, + comment=to_unset(comment), public=public, email_customer=email_customer, email_additional=email_additional, @@ -951,10 +952,10 @@ async def update_order_status( ) async def add_order_comment( context: Context, - order_id: int, + order_id: OrderIdParam, comment: Annotated[str, Field(description="Comment body")], public: Annotated[bool, Field(description="Visible to the customer")] = False, - confirm: bool = False, + confirm: ConfirmFlag = False, ) -> ToolResult: services = get_services(context) @@ -989,12 +990,12 @@ async def add_order_comment( ) async def update_order_due_date( context: Context, - order_id: int, + order_id: OrderIdParam, due_date: Annotated[str, Field(description="ISO 8601 date, e.g. '2026-03-15'")], due_date_to: Annotated[ str | None, Field(description="Optional end of the range") ] = None, - confirm: bool = False, + confirm: ConfirmFlag = False, ) -> ToolResult: services = get_services(context) @@ -1048,7 +1049,7 @@ async def bulk_update_order_status( public: bool = False, email_customer: bool = True, email_additional: bool = True, - confirm: bool = False, + confirm: ConfirmFlag = False, ) -> ToolResult: services = get_services(context) @@ -1079,7 +1080,7 @@ async def bulk_update_order_status( body = BulkStatusUpdateRequest( order_ids=order_ids, status_code=status_code, - comment=comment, + comment=to_unset(comment), public=public, email_customer=email_customer, email_additional=email_additional, diff --git a/statuspro_mcp_server/src/statuspro_mcp/tools/param_types.py b/statuspro_mcp_server/src/statuspro_mcp/tools/param_types.py new file mode 100644 index 0000000..642310f --- /dev/null +++ b/statuspro_mcp_server/src/statuspro_mcp/tools/param_types.py @@ -0,0 +1,22 @@ +"""Reusable ``Annotated[..., Field(...)]`` aliases for MCP tool parameters. + +Tool registrations push ``Field(description=...)`` metadata into the JSON +schema FastMCP advertises to clients, which gives the LLM inline guidance +about each argument. When the same parameter (``order_id``, the +``confirm`` two-step flag) appears across many tools, repeating the full +``Annotated[...]`` literal at every call site drifts: descriptions go out +of sync between siblings, the convention is harder to enforce in review, +and bare ``int`` / ``bool`` slips back in (the documented anti-pattern). + +Mirrors the pattern in ``list_coercion.py``: collapse the per-field +boilerplate into a single readable token at the call site. +""" + +from __future__ import annotations + +from typing import Annotated + +from pydantic import Field + +OrderIdParam = Annotated[int, Field(description="StatusPro order id")] +ConfirmFlag = Annotated[bool, Field(description="Set to true to apply the change")] diff --git a/statuspro_mcp_server/src/statuspro_mcp/tools/schemas.py b/statuspro_mcp_server/src/statuspro_mcp/tools/schemas.py index 07183f0..28d7347 100644 --- a/statuspro_mcp_server/src/statuspro_mcp/tools/schemas.py +++ b/statuspro_mcp_server/src/statuspro_mcp/tools/schemas.py @@ -83,7 +83,7 @@ class BatchOrderResult(BaseModel): """ order_id: int | None = Field( - None, + default=None, description="The order id if known. May be None for lookup-by-number where no id was resolved.", ) requested: str = Field( @@ -95,7 +95,7 @@ class BatchOrderResult(BaseModel): ) order: OrderSummary | None = None error: str | None = Field( - None, + default=None, description="Set when the lookup failed; describes why (not_found, ambiguous, etc.).", ) diff --git a/statuspro_mcp_server/src/statuspro_mcp/tools/statuses.py b/statuspro_mcp_server/src/statuspro_mcp/tools/statuses.py index 0fa7c9e..5d00a38 100644 --- a/statuspro_mcp_server/src/statuspro_mcp/tools/statuses.py +++ b/statuspro_mcp_server/src/statuspro_mcp/tools/statuses.py @@ -15,6 +15,7 @@ from fastmcp.tools import ToolResult from statuspro_mcp.services import get_services +from statuspro_mcp.tools.param_types import OrderIdParam from statuspro_mcp.tools.prefab_ui import build_viable_statuses_ui from statuspro_mcp.tools.schemas import StatusEntry, ViableStatusesResponse from statuspro_mcp.tools.tool_result_utils import UI_META, make_tool_result @@ -50,7 +51,10 @@ async def list_statuses(context: Context) -> list[StatusEntry]: ), meta=UI_META, ) - async def get_viable_statuses(context: Context, order_id: int) -> ToolResult: + async def get_viable_statuses( + context: Context, + order_id: OrderIdParam, + ) -> ToolResult: services = get_services(context) statuses = await services.client.statuses.viable_for(order_id) entries = [_to_entry(s) for s in statuses]