vendors.py•6.04 kB
"""Vendor management tools for B4B MCP server."""
from typing import Optional
import httpx
import structlog
from mcp.server.fastmcp import Context
from ..auth.token_handler import extract_user_token
from ..client.api_client import get_api_client
from ..server import mcp
from ..utils.validators import validate_uuid, validate_page_params
logger = structlog.get_logger(__name__)
api_client = get_api_client()
@mcp.tool()
async def list_vendors(
entity_id: str,
page: int = 1,
per_page: int = 20,
search: Optional[str] = None,
ctx: Context = None
) -> dict:
"""
List vendors for an entity.
Args:
entity_id: UUID of the entity
page: Page number (default: 1)
per_page: Items per page, max 100 (default: 20)
search: Optional search query for vendor name or tax ID
ctx: MCP context (automatically provided)
Returns:
Paginated list of vendors
"""
log = logger.bind(action="list_vendors", entity_id=entity_id, page=page, per_page=per_page)
log.info("Listing vendors")
try:
# Extract user token
token = await extract_user_token(ctx)
# Validate inputs
entity_id = validate_uuid(entity_id, "entity_id")
page, per_page = validate_page_params(page, per_page)
# Build query parameters
params = {
"page": page,
"per_page": per_page
}
if search:
params["search"] = search
# Call B4B API
response_data = await api_client.get(
f"/entities/{entity_id}/vendors",
token=token,
params=params
)
log.info(
"Vendors retrieved successfully",
total_vendors=response_data.get("total", 0),
returned_count=len(response_data.get("data", []))
)
return {
"success": True,
"vendors": response_data
}
except ValueError as e:
log.warning("Validation error", error=str(e))
return {
"success": False,
"error": f"Validation error: {str(e)}"
}
except httpx.HTTPStatusError as e:
log.error("HTTP error listing vendors", status_code=e.response.status_code)
error_detail = ""
try:
error_body = e.response.json()
error_detail = error_body.get("detail", "") or error_body.get("message", "")
except Exception:
error_detail = e.response.text
if e.response.status_code == 403:
return {
"success": False,
"error": "Access denied. You do not have permission to view vendors for this entity."
}
elif e.response.status_code == 404:
# Could be entity not found OR endpoint not implemented
return {
"success": False,
"error": f"Vendor endpoint not available: {error_detail or 'Entity not found or vendor feature not yet implemented on the backend.'}"
}
return {
"success": False,
"error": f"Failed to list vendors: {error_detail or f'HTTP {e.response.status_code}'}"
}
except Exception as e:
log.error("Unexpected error listing vendors", error=str(e), error_type=type(e).__name__)
return {
"success": False,
"error": f"Unexpected error: {str(e)}"
}
@mcp.tool()
async def get_vendor(
entity_id: str,
vendor_id: str,
ctx: Context = None
) -> dict:
"""
Get vendor details including transaction history.
Args:
entity_id: UUID of the entity
vendor_id: UUID of the vendor
ctx: MCP context (automatically provided)
Returns:
Complete vendor details with transaction history
"""
log = logger.bind(action="get_vendor", entity_id=entity_id, vendor_id=vendor_id)
log.info("Getting vendor details")
try:
# Extract user token
token = await extract_user_token(ctx)
# Validate inputs
entity_id = validate_uuid(entity_id, "entity_id")
vendor_id = validate_uuid(vendor_id, "vendor_id")
# Call B4B API
response_data = await api_client.get(
f"/entities/{entity_id}/vendors/{vendor_id}",
token=token
)
log.info(
"Vendor details retrieved successfully",
vendor_name=response_data.get("name", "Unknown")
)
return {
"success": True,
"vendor": response_data
}
except ValueError as e:
log.warning("Validation error", error=str(e))
return {
"success": False,
"error": f"Validation error: {str(e)}"
}
except httpx.HTTPStatusError as e:
log.error("HTTP error getting vendor", status_code=e.response.status_code)
error_detail = ""
try:
error_body = e.response.json()
error_detail = error_body.get("detail", "") or error_body.get("message", "")
except Exception:
error_detail = e.response.text
if e.response.status_code == 403:
return {
"success": False,
"error": "Access denied. You do not have permission to view this vendor."
}
elif e.response.status_code == 404:
# Could be vendor not found OR endpoint not implemented
return {
"success": False,
"error": f"Vendor not available: {error_detail or 'Vendor not found or vendor feature not yet implemented on the backend.'}"
}
return {
"success": False,
"error": f"Failed to get vendor: {error_detail or f'HTTP {e.response.status_code}'}"
}
except Exception as e:
log.error("Unexpected error getting vendor", error=str(e), error_type=type(e).__name__)
return {
"success": False,
"error": f"Unexpected error: {str(e)}"
}