mirror of
https://github.com/hanzoai/python-sdk.git
synced 2026-10-27 04:20:09 +00:00
zeekay
b4b0f3bf67
The base was api.hanzo.ai/api/v1 and then /v1; both 404. The surface is under /v1/commerce, and the fleet names each resource in the singular — product, collection, store, discount, webhook — so every read here asked for a plural that does not exist. Orders and users have no address at all: an order is reached through the store that holds it, and a user is IAM's. Both now say which, because a 404 reads to an agent as an empty result rather than as a wrong question.
249 lines
8.4 KiB
Python
249 lines
8.4 KiB
Python
"""MCP tool for Hanzo Commerce -- orders, products, collections, stores.
|
|
|
|
Wraps the Hanzo Commerce API at api.hanzo.ai/v1/commerce/.
|
|
Auth: Uses HanzoSession from hanzo-tools-auth for bearer tokens.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import json
|
|
import logging
|
|
from typing import Any, Annotated, final
|
|
|
|
from pydantic import Field
|
|
from mcp.server import FastMCP
|
|
from mcp.server.fastmcp import Context as MCPContext
|
|
|
|
from hanzo_tools.core.base import BaseTool
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
DESCRIPTION = """Hanzo Commerce -- orders, products, collections, stores, and discounts.
|
|
|
|
Requires authentication via `hanzo login` (stored at ~/.hanzo/auth/token.json).
|
|
|
|
Actions:
|
|
- orders: List orders (optional query param for filtering)
|
|
- order: Get order by ID (order_id required)
|
|
- products: List products
|
|
- product: Get product by ID (product_id required)
|
|
- collections: List product collections
|
|
- search_users: Search users (query required)
|
|
- search_orders: Search orders (query required)
|
|
- stores: List stores
|
|
- discounts: List discount codes
|
|
- webhooks: List configured webhooks
|
|
"""
|
|
|
|
API_BASE = "https://api.hanzo.ai/v1/commerce"
|
|
|
|
|
|
def _get_session():
|
|
"""Get HanzoSession singleton."""
|
|
from hanzo_tools.auth.session import HanzoSession
|
|
return HanzoSession.get()
|
|
|
|
|
|
def _api_base() -> str:
|
|
"""Get commerce API base URL (overridable via env)."""
|
|
return os.getenv("HANZO_COMMERCE_API_URL", API_BASE).rstrip("/")
|
|
|
|
|
|
def _request(method: str, path: str, token: str, **kwargs: Any) -> Any:
|
|
"""Make an authenticated HTTP request to the commerce API."""
|
|
import httpx
|
|
|
|
url = f"{_api_base()}{path}"
|
|
headers = {
|
|
"Authorization": f"Bearer {token}",
|
|
"Content-Type": "application/json",
|
|
"User-Agent": "hanzo-mcp/0.1",
|
|
}
|
|
|
|
with httpx.Client(timeout=30.0) as client:
|
|
resp = client.request(method, url, headers=headers, **kwargs)
|
|
if resp.status_code >= 400:
|
|
try:
|
|
err = resp.json()
|
|
msg = err.get("error", err.get("message", resp.text))
|
|
except Exception:
|
|
msg = resp.text
|
|
raise RuntimeError(f"Commerce API error {resp.status_code}: {msg}")
|
|
if not resp.content or resp.status_code == 204:
|
|
return {}
|
|
return resp.json()
|
|
|
|
|
|
def _get(path: str, token: str, params: dict[str, str] | None = None) -> Any:
|
|
return _request("GET", path, token, params=params)
|
|
|
|
|
|
@final
|
|
class CommerceTool(BaseTool):
|
|
"""MCP tool for Hanzo commerce operations."""
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return "commerce"
|
|
|
|
@property
|
|
def description(self) -> str:
|
|
return DESCRIPTION
|
|
|
|
def _get_token(self) -> str:
|
|
"""Get auth token or raise."""
|
|
session = _get_session()
|
|
token = session.get_iam_token()
|
|
if not token:
|
|
raise RuntimeError("Not authenticated. Run 'hanzo login' first.")
|
|
return token
|
|
|
|
async def call(
|
|
self,
|
|
ctx: MCPContext,
|
|
action: str = "orders",
|
|
order_id: str | None = None,
|
|
product_id: str | None = None,
|
|
query: str | None = None,
|
|
**kwargs: Any,
|
|
) -> str:
|
|
try:
|
|
if action == "orders":
|
|
return await self._orders(query)
|
|
elif action == "order":
|
|
return await self._order(order_id)
|
|
elif action == "products":
|
|
return await self._products(query)
|
|
elif action == "product":
|
|
return await self._product(product_id)
|
|
elif action == "collections":
|
|
return await self._collections()
|
|
elif action == "search_users":
|
|
return await self._search_users(query)
|
|
elif action == "search_orders":
|
|
return await self._search_orders(query)
|
|
elif action == "stores":
|
|
return await self._stores()
|
|
elif action == "discounts":
|
|
return await self._discounts()
|
|
elif action == "webhooks":
|
|
return await self._webhooks()
|
|
else:
|
|
return json.dumps({
|
|
"error": f"Unknown action: {action}",
|
|
"available": [
|
|
"orders", "order", "products", "product", "collections",
|
|
"search_users", "search_orders", "stores", "discounts", "webhooks",
|
|
],
|
|
})
|
|
except RuntimeError as e:
|
|
return json.dumps({"error": str(e)})
|
|
except Exception as e:
|
|
logger.exception(f"Commerce tool error: {e}")
|
|
return json.dumps({"error": f"Commerce error: {e}"})
|
|
|
|
# -- Actions -------------------------------------------------------------
|
|
|
|
# The fleet publishes no order collection under /v1/commerce. An order is
|
|
# reachable only through the store that holds it — authorize and capture take
|
|
# an orderid — so there is no address to list or fetch one by id. Saying so is
|
|
# the honest answer; asking anyway returned a 404 an agent read as "no orders".
|
|
_NO_ORDERS = (
|
|
"Hanzo Commerce publishes no order collection. An order is addressed "
|
|
"through its store (/v1/commerce/store/{storeid}/authorize/{orderid}), "
|
|
"so listing or fetching one by id has no endpoint."
|
|
)
|
|
|
|
async def _orders(self, query: str | None) -> str:
|
|
return json.dumps({"error": self._NO_ORDERS})
|
|
|
|
async def _order(self, order_id: str | None) -> str:
|
|
return json.dumps({"error": self._NO_ORDERS})
|
|
|
|
async def _products(self, query: str | None) -> str:
|
|
token = self._get_token()
|
|
params = {"q": query} if query else None
|
|
data = _get("/product/", token, params=params)
|
|
return json.dumps(data, indent=2)
|
|
|
|
async def _product(self, product_id: str | None) -> str:
|
|
if not product_id:
|
|
return json.dumps({"error": "Required: product_id"})
|
|
token = self._get_token()
|
|
data = _get(f"/product/{product_id}", token)
|
|
return json.dumps(data, indent=2)
|
|
|
|
async def _collections(self) -> str:
|
|
token = self._get_token()
|
|
data = _get("/collection/", token)
|
|
return json.dumps(data, indent=2)
|
|
|
|
async def _search_users(self, query: str | None) -> str:
|
|
# A user is identity, and identity is IAM's. Commerce holds subscribers
|
|
# and wallets, never people.
|
|
return json.dumps({
|
|
"error": "Users are IAM's, not commerce's — search them at /v1/iam."
|
|
})
|
|
|
|
async def _search_orders(self, query: str | None) -> str:
|
|
return json.dumps({"error": self._NO_ORDERS})
|
|
|
|
async def _stores(self) -> str:
|
|
token = self._get_token()
|
|
data = _get("/store/", token)
|
|
return json.dumps(data, indent=2)
|
|
|
|
async def _discounts(self) -> str:
|
|
token = self._get_token()
|
|
data = _get("/discount/", token)
|
|
return json.dumps(data, indent=2)
|
|
|
|
async def _webhooks(self) -> str:
|
|
token = self._get_token()
|
|
data = _get("/webhook/", token)
|
|
return json.dumps(data, indent=2)
|
|
|
|
# -- Registration --------------------------------------------------------
|
|
|
|
def register(self, mcp_server: FastMCP) -> None:
|
|
"""Register commerce tool with explicit parameters."""
|
|
tool_instance = self
|
|
|
|
@mcp_server.tool(
|
|
name="commerce",
|
|
description=DESCRIPTION,
|
|
)
|
|
async def commerce(
|
|
action: Annotated[
|
|
str,
|
|
Field(
|
|
description=(
|
|
"Action to perform. "
|
|
"orders, order, products, product, collections, "
|
|
"search_users, search_orders, stores, discounts, webhooks."
|
|
),
|
|
),
|
|
] = "orders",
|
|
order_id: Annotated[
|
|
str | None,
|
|
Field(description="Order ID (for order action)"),
|
|
] = None,
|
|
product_id: Annotated[
|
|
str | None,
|
|
Field(description="Product ID (for product action)"),
|
|
] = None,
|
|
query: Annotated[
|
|
str | None,
|
|
Field(description="Search query (for search_users, search_orders, or filtering orders/products)"),
|
|
] = None,
|
|
ctx: MCPContext = None,
|
|
) -> str:
|
|
return await tool_instance.call(
|
|
ctx,
|
|
action=action,
|
|
order_id=order_id,
|
|
product_id=product_id,
|
|
query=query,
|
|
)
|