#!/usr/bin/env python3
"""
SmartKasa MCP Server - Production-Ready Model Context Protocol Server
======================================================================
Complete MCP server for SmartKasa API providing full CRUD operations
for all API resources with enterprise-grade features.
Features:
- Async HTTP with connection pooling (httpx)
- Automatic token refresh with exponential backoff
- Structured logging with configurable levels
- Rate limiting protection
- Comprehensive error handling with retry logic
- Thread-safe session management
- Secure credential handling
Architecture Note (Multi-User Support):
---------------------------------------
MCP uses stdio transport where each LLM client spawns a SEPARATE process
of this server. This means:
- Each user/client gets their own isolated process
- Session state is automatically isolated per user
- No cross-contamination of credentials between users
- Process termination = automatic session cleanup
For persistent sessions across restarts, use environment variables
or the optional encrypted file storage.
Usage:
python smartkasa-mcp-server.py
Environment variables (optional, can also be set via tools):
SMARTKASA_API_KEY - API key for authentication
SMARTKASA_PHONE - Phone number for login
SMARTKASA_PASSWORD - Password for login
SMARTKASA_LOG_LEVEL - Logging level (DEBUG, INFO, WARNING, ERROR)
SMARTKASA_BASE_URL - API base URL (default: https://core.smartkasa.ua)
"""
import os
import sys
import json
import asyncio
import logging
import argparse
from datetime import datetime, timezone, timedelta
from typing import Optional, Dict, Any, List
from dataclasses import dataclass, field
import httpx
from mcp.server import Server
from mcp.types import Tool, TextContent
from mcp.server.stdio import stdio_server
# SSE transport imports (optional, for remote deployment)
try:
from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.routing import Route
from starlette.responses import JSONResponse
import uvicorn
SSE_AVAILABLE = True
except ImportError:
SSE_AVAILABLE = False
# =============================================================================
# CONFIGURATION
# =============================================================================
@dataclass
class Config:
"""Server configuration with sensible defaults."""
base_url: str = field(default_factory=lambda: os.getenv(
"SMARTKASA_BASE_URL", "https://core.smartkasa.ua"))
log_level: str = field(default_factory=lambda: os.getenv(
"SMARTKASA_LOG_LEVEL", "INFO"))
transport: str = field(default_factory=lambda: os.getenv(
"SMARTKASA_TRANSPORT", "stdio")) # stdio or sse
sse_host: str = field(default_factory=lambda: os.getenv(
"SMARTKASA_SSE_HOST", "0.0.0.0"))
sse_port: int = field(default_factory=lambda: int(
os.getenv("SMARTKASA_SSE_PORT", "8080")))
request_timeout: float = 30.0
max_retries: int = 3
retry_delay: float = 1.0
connection_pool_size: int = 10
keepalive_expiry: float = 30.0
token_refresh_margin: int = 300 # Refresh 5 min before expiry
config = Config()
# =============================================================================
# LOGGING SETUP
# =============================================================================
def setup_logging() -> logging.Logger:
"""Configure structured logging to stderr (stdout reserved for MCP)."""
logger = logging.getLogger("smartkasa-mcp")
logger.setLevel(getattr(logging, config.log_level.upper(), logging.INFO))
# Only stderr - stdout is for MCP protocol
handler = logging.StreamHandler(sys.stderr)
handler.setFormatter(logging.Formatter(
"%(asctime)s | %(levelname)-8s | %(message)s",
datefmt="%Y-%m-%d %H:%M:%S"
))
if not logger.handlers:
logger.addHandler(handler)
return logger
logger = setup_logging()
# =============================================================================
# SESSION MANAGEMENT (Per-Process Isolation)
# =============================================================================
@dataclass
class SessionState:
"""
Thread-safe session state for a single user.
Security Model:
- Credentials stored in memory only (not persisted to disk by default)
- Passwords cleared from memory after authentication
- Tokens auto-expire and require re-authentication
- Process isolation ensures no credential leakage between users
"""
# Credentials (sensitive)
api_key: Optional[str] = None
phone_number: Optional[str] = None
password: Optional[str] = None # Cleared after successful auth
# Tokens
access_token: Optional[str] = None
refresh_token: Optional[str] = None
access_expires_at: Optional[datetime] = None
refresh_expires_at: Optional[datetime] = None
# Session metadata
authenticated_at: Optional[datetime] = None
last_request_at: Optional[datetime] = None
request_count: int = 0
def is_token_expired(self) -> bool:
"""Check if access token is expired or about to expire."""
if not self.access_expires_at:
return True
margin = timedelta(seconds=config.token_refresh_margin)
return datetime.now(timezone.utc) >= (self.access_expires_at - margin)
def is_refresh_token_valid(self) -> bool:
"""Check if refresh token is still valid."""
if not self.refresh_expires_at:
return False
return datetime.now(timezone.utc) < self.refresh_expires_at
def clear_tokens(self) -> None:
"""Clear all tokens (for logout)."""
self.access_token = None
self.refresh_token = None
self.access_expires_at = None
self.refresh_expires_at = None
self.authenticated_at = None
def clear_password(self) -> None:
"""Clear password from memory after successful auth."""
self.password = None
def get_auth_status(self) -> Dict[str, Any]:
"""Get current authentication status (safe for logging)."""
return {
"has_api_key": bool(self.api_key),
"has_credentials": bool(self.phone_number),
"is_authenticated": bool(self.access_token),
"token_expired": self.is_token_expired() if self.access_token else None,
"expires_at": self.access_expires_at.isoformat() if self.access_expires_at else None,
"authenticated_at": self.authenticated_at.isoformat() if self.authenticated_at else None,
"request_count": self.request_count
}
# Global session state (isolated per process)
session = SessionState(
api_key=os.getenv("SMARTKASA_API_KEY"),
phone_number=os.getenv("SMARTKASA_PHONE"),
password=os.getenv("SMARTKASA_PASSWORD")
)
# =============================================================================
# HTTP CLIENT WITH CONNECTION POOLING
# =============================================================================
class SmartKasaClient:
"""
High-performance async HTTP client with:
- Connection pooling for efficiency
- Automatic retry with exponential backoff
- Token refresh handling
- Request/response logging
"""
def __init__(self):
self._client: Optional[httpx.AsyncClient] = None
async def get_client(self) -> httpx.AsyncClient:
"""Get or create the HTTP client with connection pooling."""
if self._client is None or self._client.is_closed:
self._client = httpx.AsyncClient(
base_url=config.base_url,
timeout=httpx.Timeout(config.request_timeout),
limits=httpx.Limits(
max_connections=config.connection_pool_size,
max_keepalive_connections=config.connection_pool_size,
keepalive_expiry=config.keepalive_expiry
),
http2=True # Enable HTTP/2 for better performance
)
logger.debug(
f"Created HTTP client with pool size {config.connection_pool_size}")
return self._client
async def close(self) -> None:
"""Close the HTTP client."""
if self._client and not self._client.is_closed:
await self._client.aclose()
logger.debug("HTTP client closed")
def _get_headers(self, include_auth: bool = True) -> Dict[str, str]:
"""Build request headers."""
headers = {
"Content-Type": "application/json",
"Accept": "application/json",
"User-Agent": "SmartKasa-MCP-Server/1.0"
}
if session.api_key:
headers["X-Api-Key"] = session.api_key
if include_auth and session.access_token:
headers["Authorization"] = f"Bearer {session.access_token}"
return headers
async def _authenticate(self) -> Dict[str, Any]:
"""Authenticate with SmartKasa API."""
if not all([session.api_key, session.phone_number, session.password]):
return {
"success": False,
"error": "Missing credentials. Use smartkasa_set_credentials first.",
"required": ["api_key", "phone_number", "password"]
}
payload = {
"session": {
"phone_number": session.phone_number,
"password": session.password
}
}
client = await self.get_client()
try:
response = await client.post(
"/api/v1/auth/sessions",
headers=self._get_headers(include_auth=False),
json=payload
)
if response.status_code == 201:
data = response.json()["data"]
session.access_token = data["access"]
session.refresh_token = data["refresh"]
session.access_expires_at = datetime.fromisoformat(
data["access_expires_at"].replace("Z", "+00:00")
)
session.refresh_expires_at = datetime.fromisoformat(
data["refresh_expires_at"].replace("Z", "+00:00")
)
session.authenticated_at = datetime.now(timezone.utc)
# Security: clear password from memory after successful auth
# (can be re-set if needed for re-authentication)
# session.clear_password() # Uncomment for maximum security
logger.info(
f"Authentication successful, expires at {session.access_expires_at}")
return {
"success": True,
"message": "Authentication successful",
"expires_at": session.access_expires_at.isoformat()
}
else:
logger.warning(
f"Authentication failed: {response.status_code}")
return {
"success": False,
"error": f"Authentication failed: {response.status_code}",
"details": response.text
}
except httpx.RequestError as e:
logger.error(f"Authentication request error: {e}")
return {"success": False, "error": f"Request error: {str(e)}"}
async def _refresh_token(self) -> Dict[str, Any]:
"""Refresh the access token using refresh token."""
if not session.is_refresh_token_valid():
logger.info("Refresh token invalid, re-authenticating")
return await self._authenticate()
client = await self.get_client()
headers = self._get_headers(include_auth=False)
headers["Authorization"] = f"Bearer {session.refresh_token}"
try:
response = await client.post("/api/v1/auth/refresh", headers=headers)
if response.status_code == 201:
data = response.json()["data"]
session.access_token = data["access"]
session.refresh_token = data["refresh"]
session.access_expires_at = datetime.fromisoformat(
data["access_expires_at"].replace("Z", "+00:00")
)
session.refresh_expires_at = datetime.fromisoformat(
data["refresh_expires_at"].replace("Z", "+00:00")
)
logger.debug("Token refreshed successfully")
return {"success": True, "message": "Token refreshed"}
else:
logger.info("Token refresh failed, re-authenticating")
return await self._authenticate()
except httpx.RequestError as e:
logger.error(f"Token refresh error: {e}")
return await self._authenticate()
async def ensure_authenticated(self) -> Optional[str]:
"""Ensure we have a valid access token."""
if not session.access_token or session.is_token_expired():
result = await self._refresh_token()
if not result["success"]:
return result.get("error", "Authentication failed")
return None
async def request(
self,
method: str,
endpoint: str,
params: Optional[Dict] = None,
data: Optional[Dict] = None,
retry_count: int = 0
) -> Dict[str, Any]:
"""
Make an authenticated API request with retry logic.
Features:
- Automatic token refresh on 401
- Exponential backoff on failures
- Request counting for monitoring
"""
# Ensure authentication
auth_error = await self.ensure_authenticated()
if auth_error:
return {"success": False, "error": auth_error}
client = await self.get_client()
session.last_request_at = datetime.now(timezone.utc)
session.request_count += 1
try:
response = await client.request(
method=method,
url=endpoint,
headers=self._get_headers(),
params=params,
json=data
)
# Success
if response.status_code in [200, 201]:
return {"success": True, "data": response.json()}
# No content (successful delete)
if response.status_code == 204:
return {"success": True, "data": {"deleted": True}}
# Unauthorized - try to refresh and retry
if response.status_code == 401 and retry_count < 1:
logger.info("Got 401, attempting token refresh")
await self._authenticate()
return await self.request(method, endpoint, params, data, retry_count + 1)
# Rate limited - wait and retry
if response.status_code == 429 and retry_count < config.max_retries:
retry_after = int(response.headers.get(
"Retry-After", config.retry_delay * (2 ** retry_count)))
logger.warning(f"Rate limited, waiting {retry_after}s")
await asyncio.sleep(retry_after)
return await self.request(method, endpoint, params, data, retry_count + 1)
# Server error - retry with backoff
if response.status_code >= 500 and retry_count < config.max_retries:
delay = config.retry_delay * (2 ** retry_count)
logger.warning(
f"Server error {response.status_code}, retrying in {delay}s")
await asyncio.sleep(delay)
return await self.request(method, endpoint, params, data, retry_count + 1)
# Final failure
logger.error(
f"Request failed: {method} {endpoint} -> {response.status_code}")
return {
"success": False,
"error": f"API request failed: {response.status_code}",
"details": response.text
}
except httpx.TimeoutException:
if retry_count < config.max_retries:
delay = config.retry_delay * (2 ** retry_count)
logger.warning(f"Request timeout, retrying in {delay}s")
await asyncio.sleep(delay)
return await self.request(method, endpoint, params, data, retry_count + 1)
return {"success": False, "error": "Request timeout after retries"}
except httpx.RequestError as e:
logger.error(f"Request error: {e}")
return {"success": False, "error": f"Request error: {str(e)}"}
# Global HTTP client (reused across requests)
http_client = SmartKasaClient()
# =============================================================================
# MCP SERVER SETUP
# =============================================================================
server = Server("smartkasa-mcp")
# =============================================================================
# TOOL DEFINITIONS
# =============================================================================
def create_tool(name: str, description: str, properties: Dict, required: List[str] = None) -> Tool:
"""Helper to create a Tool with consistent structure."""
schema = {"type": "object", "properties": properties}
if required:
schema["required"] = required
return Tool(name=name, description=description, inputSchema=schema)
@server.list_tools()
async def list_tools():
"""List all available SmartKasa tools."""
return [
# ==================== AUTHENTICATION ====================
create_tool(
"smartkasa_set_credentials",
"Set SmartKasa API credentials. Must be called before any other operations. Credentials are stored in memory only for this session.",
{
"api_key": {"type": "string", "description": "SmartKasa API key (X-Api-Key header)"},
"phone_number": {"type": "string", "description": "Phone number (e.g., 380501234567)"},
"password": {"type": "string", "description": "Password for authentication"}
},
["api_key", "phone_number", "password"]
),
create_tool(
"smartkasa_authenticate",
"Authenticate with SmartKasa API using stored credentials. Returns access token expiration time.",
{}
),
create_tool(
"smartkasa_get_status",
"Get current authentication status including token expiration, request count, and session info.",
{}
),
create_tool(
"smartkasa_logout",
"Logout and invalidate current session. Clears all tokens from memory.",
{}
),
# ==================== TERMINALS ====================
create_tool(
"smartkasa_terminals_list",
"Get list of POS terminals with optional filtering.",
{
"id": {"type": "integer", "description": "Filter by terminal ID"},
"name": {"type": "string", "description": "Filter by terminal name"},
"shop_id": {"type": "integer", "description": "Filter by shop ID"},
"page": {"type": "integer", "description": "Page number"},
"per_page": {"type": "integer", "description": "Items per page (default: 25)"}
}
),
create_tool(
"smartkasa_terminals_get",
"Get single terminal by ID.",
{"id": {"type": "integer", "description": "Terminal ID"}},
["id"]
),
create_tool(
"smartkasa_terminals_update",
"Update terminal configuration.",
{
"id": {"type": "integer", "description": "Terminal ID"},
"name": {"type": "string", "description": "New terminal name"},
"shop_id": {"type": "integer", "description": "New shop ID"}
},
["id"]
),
create_tool(
"smartkasa_terminals_delete",
"Delete a terminal.",
{"id": {"type": "integer", "description": "Terminal ID to delete"}},
["id"]
),
# ==================== SHOPS ====================
create_tool(
"smartkasa_shops_list",
"Get list of shops (retail locations).",
{
"title": {"type": "string", "description": "Filter by shop name"},
"state": {"type": "integer", "description": "Filter by state (0=active)"},
"fiscalization_enabled": {"type": "boolean", "description": "Filter by fiscalization"},
"page": {"type": "integer", "description": "Page number"},
"per_page": {"type": "integer", "description": "Items per page"}
}
),
create_tool(
"smartkasa_shops_get",
"Get single shop by ID.",
{"id": {"type": "integer", "description": "Shop ID"}},
["id"]
),
create_tool(
"smartkasa_shops_create",
"Create a new shop location.",
{
"title": {"type": "string", "description": "Shop name"},
"email": {"type": "string", "description": "Email address"},
"phone_number": {"type": "string", "description": "Phone number"},
"website_url": {"type": "string", "description": "Website URL"},
"business_category_id": {"type": "integer", "description": "Business category ID"},
"fiscalization_enabled": {"type": "boolean", "description": "Enable fiscalization"},
"is_vat_taxation": {"type": "boolean", "description": "VAT payer flag"},
"is_excise_taxation": {"type": "boolean", "description": "Excise taxation flag"},
"address": {"type": "object", "description": "Address object with zip_code, city_name, content"}
},
["title"]
),
create_tool(
"smartkasa_shops_update",
"Update shop data.",
{
"id": {"type": "integer", "description": "Shop ID"},
"title": {"type": "string", "description": "Shop name"},
"email": {"type": "string", "description": "Email"},
"phone_number": {"type": "string", "description": "Phone"},
"website_url": {"type": "string", "description": "Website"},
"fiscalization_enabled": {"type": "boolean", "description": "Fiscalization flag"},
"is_vat_taxation": {"type": "boolean", "description": "VAT flag"},
"address": {"type": "object", "description": "Address object"}
},
["id"]
),
create_tool(
"smartkasa_shops_delete",
"Delete a shop.",
{"id": {"type": "integer", "description": "Shop ID to delete"}},
["id"]
),
create_tool(
"smartkasa_shops_employees",
"Get list of employees across all shops.",
{}
),
# ==================== EMPLOYEES ====================
create_tool(
"smartkasa_employees_list",
"Get list of employees with filtering.",
{
"first_name": {"type": "string", "description": "Filter by first name"},
"last_name": {"type": "string", "description": "Filter by last name"},
"phone_number": {"type": "string", "description": "Filter by phone"},
"email": {"type": "string", "description": "Filter by email"},
"role_id": {"type": "integer", "description": "Filter by role ID"},
"page": {"type": "integer", "description": "Page number"},
"per_page": {"type": "integer", "description": "Items per page"}
}
),
create_tool(
"smartkasa_employees_get",
"Get single employee by ID.",
{"id": {"type": "integer", "description": "Employee ID"}},
["id"]
),
create_tool(
"smartkasa_employees_create",
"Create a new employee.",
{
"first_name": {"type": "string", "description": "First name"},
"last_name": {"type": "string", "description": "Last name"},
"middle_name": {"type": "string", "description": "Middle name (patronymic)"},
"email": {"type": "string", "description": "Email"},
"phone_number": {"type": "string", "description": "Phone number"},
"inn": {"type": "string", "description": "Tax ID (ІПН)"},
"pin_code": {"type": "string", "description": "POS access PIN code"},
"role_id": {"type": "integer", "description": "Role/position ID"},
"gender_type_id": {"type": "integer", "description": "Gender (0=female, 1=male)"},
"birthday": {"type": "string", "description": "Birthday (YYYY-MM-DD)"},
"joined_at": {"type": "string", "description": "Employment start date"}
},
["first_name", "last_name", "phone_number"]
),
create_tool(
"smartkasa_employees_update",
"Update employee profile.",
{
"id": {"type": "integer", "description": "Employee ID"},
"first_name": {"type": "string", "description": "First name"},
"last_name": {"type": "string", "description": "Last name"},
"email": {"type": "string", "description": "Email"},
"phone_number": {"type": "string", "description": "Phone"},
"pin_code": {"type": "string", "description": "PIN code"},
"role_id": {"type": "integer", "description": "Role ID"}
},
["id"]
),
create_tool(
"smartkasa_employees_delete",
"Delete an employee.",
{"id": {"type": "integer", "description": "Employee ID to delete"}},
["id"]
),
# ==================== UNIT TYPES ====================
create_tool(
"smartkasa_unit_types_list",
"Get list of available unit types (штука, кг, година, etc.).",
{}
),
# ==================== CATEGORIES ====================
create_tool(
"smartkasa_categories_list",
"Get list of product categories.",
{
"title": {"type": "string", "description": "Filter by category name"},
"parent_id": {"type": "string", "description": "Filter by parent category ID (UUID)"},
"color_id": {"type": "integer", "description": "Filter by color"},
"page": {"type": "integer", "description": "Page number"},
"per_page": {"type": "integer", "description": "Items per page"}
}
),
create_tool(
"smartkasa_categories_create",
"Create a new product category.",
{
"title": {"type": "string", "description": "Category name"},
"parent_id": {"type": "string", "description": "Parent category ID (UUID, null for root)"},
"color_id": {"type": "integer", "description": "Color identifier"},
"shop_ids": {"type": "array", "description": "Array of shop IDs", "items": {"type": "integer"}}
},
["title"]
),
create_tool(
"smartkasa_categories_update",
"Update a category.",
{
"id": {"type": "string", "description": "Category ID (UUID)"},
"title": {"type": "string", "description": "Category name"},
"parent_id": {"type": "string", "description": "Parent category ID"},
"color_id": {"type": "integer", "description": "Color ID"}
},
["id"]
),
create_tool(
"smartkasa_categories_delete",
"Delete a category.",
{"id": {"type": "string",
"description": "Category ID (UUID) to delete"}},
["id"]
),
create_tool(
"smartkasa_categories_batch_create",
"Batch create multiple categories.",
{
"categories": {
"type": "array",
"description": "Array of category objects with title and color",
"items": {"type": "object"}
}
},
["categories"]
),
create_tool(
"smartkasa_categories_batch_delete",
"Batch delete multiple categories.",
{
"ids": {
"type": "array",
"description": "Array of category IDs (UUIDs) to delete",
"items": {"type": "string"}
}
},
["ids"]
),
# ==================== PRODUCTS ====================
create_tool(
"smartkasa_products_list",
"Get list of products with search and filtering.",
{
"q": {"type": "string", "description": "Full-text search query"},
"number": {"type": "string", "description": "Filter by barcode"},
"category_id": {"type": "string", "description": "Filter by category ID (UUID)"},
"shop_id": {"type": "integer", "description": "Filter by shop ID"},
"subgroup_id": {"type": "integer", "description": "Filter by subgroup ID"},
"page": {"type": "integer", "description": "Page number"},
"per_page": {"type": "integer", "description": "Items per page"}
}
),
create_tool(
"smartkasa_products_get",
"Get single product by ID with inventory cards.",
{"id": {"type": "string", "description": "Product ID (UUID)"}},
["id"]
),
create_tool(
"smartkasa_products_create",
"Create a new product.",
{
"title": {"type": "string", "description": "Product name"},
"number": {"type": "string", "description": "Barcode"},
"price": {"type": "number", "description": "Price per unit"},
"procurement_price": {"type": "number", "description": "Purchase price"},
"unit_type_id": {"type": "integer", "description": "Unit type ID (0=шт, 9=кг)"},
"category_id": {"type": "string", "description": "Category ID (UUID)"},
"tax_group_id": {"type": "integer", "description": "Tax group (0-10)"},
"classifier_type_id": {"type": "integer", "description": "Classifier type (0=none, 1=УКТЗЕД, 2=ДКПП)"},
"classifier_code": {"type": "string", "description": "Classifier code"},
"sold_by_weight": {"type": "boolean", "description": "Sold by weight flag"},
"is_free_price": {"type": "boolean", "description": "Free price flag"},
"alter_title": {"type": "string", "description": "Short name"},
"alter_number": {"type": "string", "description": "SKU/Article"}
},
["title", "price"]
),
create_tool(
"smartkasa_products_update",
"Update a product (partial update with PATCH).",
{
"id": {"type": "string", "description": "Product ID (UUID)"},
"title": {"type": "string", "description": "Product name"},
"price": {"type": "number", "description": "Price"},
"category_id": {"type": "string", "description": "Category ID"},
"tax_group_id": {"type": "integer", "description": "Tax group"},
"sold_by_weight": {"type": "boolean", "description": "Sold by weight"}
},
["id"]
),
create_tool(
"smartkasa_products_delete",
"Delete a product.",
{"id": {"type": "string",
"description": "Product ID (UUID) to delete"}},
["id"]
),
create_tool(
"smartkasa_products_batch_create",
"Batch create multiple products.",
{
"products": {
"type": "array",
"description": "Array of product objects",
"items": {"type": "object"}
}
},
["products"]
),
create_tool(
"smartkasa_products_batch_delete",
"Batch delete multiple products.",
{
"ids": {
"type": "array",
"description": "Array of product IDs (UUIDs) to delete",
"items": {"type": "string"}
}
},
["ids"]
),
# ==================== INVENTORY CARDS ====================
create_tool(
"smartkasa_cards_list",
"Get list of inventory cards (stock per shop).",
{
"product_id": {"type": "string", "description": "Filter by product ID (UUID)"},
"shop_id": {"type": "integer", "description": "Filter by shop ID"},
"counting_enabled": {"type": "boolean", "description": "Filter by stock tracking"},
"page": {"type": "integer", "description": "Page number"},
"per_page": {"type": "integer", "description": "Items per page"}
}
),
create_tool(
"smartkasa_cards_get",
"Get single inventory card by ID.",
{"id": {"type": "integer", "description": "Card ID"}},
["id"]
),
create_tool(
"smartkasa_cards_create",
"Create an inventory card (assign product to shop with stock).",
{
"product_id": {"type": "string", "description": "Product ID (UUID)"},
"shop_id": {"type": "integer", "description": "Shop ID"},
"count": {"type": "number", "description": "Initial stock quantity"},
"counting_enabled": {"type": "boolean", "description": "Enable stock tracking"}
},
["product_id", "shop_id"]
),
create_tool(
"smartkasa_cards_update",
"Update inventory card (change stock).",
{
"id": {"type": "integer", "description": "Card ID"},
"count": {"type": "number", "description": "New stock quantity"},
"counting_enabled": {"type": "boolean", "description": "Stock tracking flag"}
},
["id"]
),
create_tool(
"smartkasa_cards_delete",
"Delete an inventory card.",
{"id": {"type": "integer", "description": "Card ID to delete"}},
["id"]
),
# ==================== PRODUCT SUBGROUPS ====================
create_tool(
"smartkasa_subgroups_list",
"Get list of product subgroups.",
{
"title": {"type": "string", "description": "Filter by subgroup name"},
"page": {"type": "integer", "description": "Page number"},
"per_page": {"type": "integer", "description": "Items per page"}
}
),
create_tool(
"smartkasa_subgroups_create",
"Create a product subgroup.",
{"title": {"type": "string", "description": "Subgroup name"}},
["title"]
),
create_tool(
"smartkasa_subgroups_update",
"Update a product subgroup.",
{
"id": {"type": "integer", "description": "Subgroup ID"},
"title": {"type": "string", "description": "New subgroup name"}
},
["id", "title"]
),
create_tool(
"smartkasa_subgroups_delete",
"Delete a product subgroup.",
{"id": {"type": "integer", "description": "Subgroup ID to delete"}},
["id"]
),
# ==================== IMPORT ====================
create_tool(
"smartkasa_import_list",
"Get import history.",
{
"page": {"type": "integer", "description": "Page number"},
"per_page": {"type": "integer", "description": "Items per page"}
}
),
create_tool(
"smartkasa_import_start",
"Start product import from file URL.",
{
"url": {"type": "string", "description": "File URL to import from"},
"category_id": {"type": "string", "description": "Target category ID (UUID)"}
},
["url"]
),
create_tool(
"smartkasa_import_status",
"Get import job status by ID.",
{"id": {"type": "integer", "description": "Import job ID"}},
["id"]
),
# ==================== SHIFTS ====================
create_tool(
"smartkasa_shifts_list",
"Get list of fiscal shifts.",
{
"terminal_id": {"type": "integer", "description": "Filter by terminal ID"},
"state": {"type": "integer", "description": "Filter by state (0=open, 1=closed, 2=archived)"},
"employee_id": {"type": "integer", "description": "Filter by employee ID"},
"page": {"type": "integer", "description": "Page number"},
"per_page": {"type": "integer", "description": "Items per page"}
}
),
create_tool(
"smartkasa_shifts_get",
"Get single shift by ID.",
{"id": {"type": "string", "description": "Shift ID (UUID)"}},
["id"]
),
# ==================== RECEIPTS ====================
create_tool(
"smartkasa_receipts_list",
"Get list of receipts (sales transactions).",
{
"shop_id": {"type": "integer", "description": "Filter by shop ID"},
"terminal_id": {"type": "integer", "description": "Filter by terminal ID"},
"shift_id": {"type": "string", "description": "Filter by shift ID (UUID)"},
"state": {"type": "integer", "description": "Filter by state (1=draft, 2=completed, 3=refund)"},
"type": {"type": "integer", "description": "Filter by type (0=sale, 1=service_in, 2=service_out, 3=refund)"},
"fiscal_number": {"type": "string", "description": "Filter by fiscal number"},
"date_start": {"type": "string", "description": "Start date (ISO 8601)"},
"date_end": {"type": "string", "description": "End date (ISO 8601)"},
"page": {"type": "integer", "description": "Page number"},
"per_page": {"type": "integer", "description": "Items per page"}
}
),
create_tool(
"smartkasa_receipts_get",
"Get single receipt by ID with items and transactions.",
{"id": {"type": "string", "description": "Receipt ID (UUID)"}},
["id"]
),
create_tool(
"smartkasa_receipts_create",
"Create a new receipt.",
{
"title": {"type": "string", "description": "Receipt title"},
"total_amount": {"type": "number", "description": "Total amount"},
"state": {"type": "integer", "description": "State (1=draft, 2=completed)"},
"type": {"type": "integer", "description": "Type (0=sale, 3=refund)"},
"shift_id": {"type": "string", "description": "Shift ID (UUID)"},
"shop_id": {"type": "integer", "description": "Shop ID"},
"cashier_name": {"type": "string", "description": "Cashier name"},
"discount_amount": {"type": "number", "description": "Discount amount"},
"discount_type_id": {"type": "integer", "description": "Discount type (0=%, 1=money)"},
"items": {
"type": "array",
"description": "Array of receipt items",
"items": {"type": "object"}
}
},
["total_amount", "items"]
),
create_tool(
"smartkasa_receipts_update",
"Update receipt data.",
{
"id": {"type": "string", "description": "Receipt ID (UUID)"},
"title": {"type": "string", "description": "Receipt title"},
"state": {"type": "integer", "description": "State"},
"discount_amount": {"type": "number", "description": "Discount"}
},
["id"]
),
create_tool(
"smartkasa_receipts_delete",
"Delete a receipt.",
{"id": {"type": "string",
"description": "Receipt ID (UUID) to delete"}},
["id"]
),
create_tool(
"smartkasa_receipts_batch_create",
"Batch create multiple receipts.",
{
"receipts": {
"type": "array",
"description": "Array of receipt objects",
"items": {"type": "object"}
}
},
["receipts"]
),
create_tool(
"smartkasa_receipts_batch_delete",
"Batch delete multiple receipts.",
{
"ids": {
"type": "array",
"description": "Array of receipt IDs (UUIDs) to delete",
"items": {"type": "string"}
}
},
["ids"]
),
# ==================== PAYMENT TRANSACTIONS ====================
create_tool(
"smartkasa_transactions_get",
"Get payment transaction by ID.",
{"id": {"type": "string", "description": "Transaction ID (UUID)"}},
["id"]
),
# ==================== REPORTS ====================
create_tool(
"smartkasa_reports_z_reports",
"Get Z-reports (daily fiscal closure reports).",
{
"date_start": {"type": "string", "description": "Start date (ISO 8601)"},
"date_end": {"type": "string", "description": "End date (ISO 8601)"},
"shop_id": {"type": "integer", "description": "Filter by shop ID"},
"terminal_id": {"type": "integer", "description": "Filter by terminal ID"},
"terminal_user_id": {"type": "integer", "description": "Filter by cashier ID"},
"page": {"type": "integer", "description": "Page number"},
"per_page": {"type": "integer", "description": "Items per page"}
}
),
create_tool(
"smartkasa_reports_product_sales",
"Get daily product sales statistics.",
{
"date_start": {"type": "string", "description": "Start date (ISO 8601)"},
"date_end": {"type": "string", "description": "End date (ISO 8601)"},
"shop_id": {"type": "integer", "description": "Filter by shop ID"}
},
["date_start", "date_end"]
),
]
# =============================================================================
# TOOL HANDLERS
# =============================================================================
@server.call_tool()
async def call_tool(name: str, arguments: Dict[str, Any]):
"""Handle tool calls with comprehensive error handling."""
def json_response(data: Any) -> List[TextContent]:
"""Format response as JSON text content."""
return [TextContent(type="text", text=json.dumps(data, ensure_ascii=False, indent=2, default=str))]
def filter_params(args: Dict) -> Dict:
"""Filter out None values from arguments."""
return {k: v for k, v in args.items() if v is not None}
logger.debug(f"Tool call: {name}")
try:
# ==================== AUTHENTICATION ====================
if name == "smartkasa_set_credentials":
session.api_key = arguments["api_key"]
session.phone_number = arguments["phone_number"]
session.password = arguments["password"]
logger.info("Credentials set successfully")
return json_response({
"success": True,
"message": "Credentials stored securely in memory. Call smartkasa_authenticate to login."
})
elif name == "smartkasa_authenticate":
result = await http_client._authenticate()
return json_response(result)
elif name == "smartkasa_get_status":
return json_response(session.get_auth_status())
elif name == "smartkasa_logout":
result = await http_client.request("DELETE", "/api/v1/auth/sessions/logout")
session.clear_tokens()
logger.info("Logged out successfully")
return json_response({"success": True, "message": "Logged out and tokens cleared"})
# ==================== TERMINALS ====================
elif name == "smartkasa_terminals_list":
return json_response(await http_client.request("GET", "/api/v1/pos/terminals", params=filter_params(arguments)))
elif name == "smartkasa_terminals_get":
return json_response(await http_client.request("GET", f"/api/v1/pos/terminals/{arguments['id']}"))
elif name == "smartkasa_terminals_update":
tid = arguments.pop("id")
return json_response(await http_client.request("PUT", f"/api/v1/pos/terminals/{tid}", data={"terminal": filter_params(arguments)}))
elif name == "smartkasa_terminals_delete":
return json_response(await http_client.request("DELETE", f"/api/v1/pos/terminals/{arguments['id']}"))
# ==================== SHOPS ====================
elif name == "smartkasa_shops_list":
return json_response(await http_client.request("GET", "/api/v1/rsn/shops", params=filter_params(arguments)))
elif name == "smartkasa_shops_get":
return json_response(await http_client.request("GET", f"/api/v1/rsn/shops/{arguments['id']}"))
elif name == "smartkasa_shops_create":
return json_response(await http_client.request("POST", "/api/v1/rsn/shops", data={"shop": filter_params(arguments)}))
elif name == "smartkasa_shops_update":
sid = arguments.pop("id")
return json_response(await http_client.request("PUT", f"/api/v1/rsn/shops/{sid}", data={"shop": filter_params(arguments)}))
elif name == "smartkasa_shops_delete":
return json_response(await http_client.request("DELETE", f"/api/v1/rsn/shops/{arguments['id']}"))
elif name == "smartkasa_shops_employees":
return json_response(await http_client.request("GET", "/api/v1/rsn/shops/employees"))
# ==================== EMPLOYEES ====================
elif name == "smartkasa_employees_list":
return json_response(await http_client.request("GET", "/api/v1/rsn/employees", params=filter_params(arguments)))
elif name == "smartkasa_employees_get":
return json_response(await http_client.request("GET", f"/api/v1/rsn/employees/{arguments['id']}"))
elif name == "smartkasa_employees_create":
return json_response(await http_client.request("POST", "/api/v1/rsn/employees", data={"employee": filter_params(arguments)}))
elif name == "smartkasa_employees_update":
eid = arguments.pop("id")
return json_response(await http_client.request("PUT", f"/api/v1/rsn/employees/{eid}", data={"employee": filter_params(arguments)}))
elif name == "smartkasa_employees_delete":
return json_response(await http_client.request("DELETE", f"/api/v1/rsn/employees/{arguments['id']}"))
# ==================== UNIT TYPES ====================
elif name == "smartkasa_unit_types_list":
return json_response(await http_client.request("GET", "/api/v1/inventory/unit_types"))
# ==================== CATEGORIES ====================
elif name == "smartkasa_categories_list":
return json_response(await http_client.request("GET", "/api/v1/inventory/categories", params=filter_params(arguments)))
elif name == "smartkasa_categories_create":
return json_response(await http_client.request("POST", "/api/v1/inventory/categories", data={"category": filter_params(arguments)}))
elif name == "smartkasa_categories_update":
cid = arguments.pop("id")
return json_response(await http_client.request("PUT", f"/api/v1/inventory/categories/{cid}", data={"category": filter_params(arguments)}))
elif name == "smartkasa_categories_delete":
return json_response(await http_client.request("DELETE", f"/api/v1/inventory/categories/{arguments['id']}"))
elif name == "smartkasa_categories_batch_create":
return json_response(await http_client.request("POST", "/api/v1/inventory/categories/batch", data={"batch": {"categories": arguments["categories"]}}))
elif name == "smartkasa_categories_batch_delete":
return json_response(await http_client.request("DELETE", "/api/v1/inventory/categories/batch", data={"batch": {"ids": arguments["ids"]}}))
# ==================== PRODUCTS ====================
elif name == "smartkasa_products_list":
return json_response(await http_client.request("GET", "/api/v1/inventory/products", params=filter_params(arguments)))
elif name == "smartkasa_products_get":
return json_response(await http_client.request("GET", f"/api/v1/inventory/products/{arguments['id']}"))
elif name == "smartkasa_products_create":
return json_response(await http_client.request("POST", "/api/v1/inventory/products", data={"product": filter_params(arguments)}))
elif name == "smartkasa_products_update":
pid = arguments.pop("id")
return json_response(await http_client.request("PATCH", f"/api/v1/inventory/products/{pid}", data={"product": filter_params(arguments)}))
elif name == "smartkasa_products_delete":
return json_response(await http_client.request("DELETE", f"/api/v1/inventory/products/{arguments['id']}"))
elif name == "smartkasa_products_batch_create":
return json_response(await http_client.request("POST", "/api/v1/inventory/products/batch", data={"products": arguments["products"]}))
elif name == "smartkasa_products_batch_delete":
return json_response(await http_client.request("DELETE", "/api/v1/inventory/products/batch", data={"batch": {"ids": arguments["ids"]}}))
# ==================== INVENTORY CARDS ====================
elif name == "smartkasa_cards_list":
return json_response(await http_client.request("GET", "/api/v1/inventory/cards", params=filter_params(arguments)))
elif name == "smartkasa_cards_get":
return json_response(await http_client.request("GET", f"/api/v1/inventory/cards/{arguments['id']}"))
elif name == "smartkasa_cards_create":
return json_response(await http_client.request("POST", "/api/v1/inventory/cards", data={"card": filter_params(arguments)}))
elif name == "smartkasa_cards_update":
cid = arguments.pop("id")
return json_response(await http_client.request("PUT", f"/api/v1/inventory/cards/{cid}", data={"card": filter_params(arguments)}))
elif name == "smartkasa_cards_delete":
return json_response(await http_client.request("DELETE", f"/api/v1/inventory/cards/{arguments['id']}"))
# ==================== PRODUCT SUBGROUPS ====================
elif name == "smartkasa_subgroups_list":
return json_response(await http_client.request("GET", "/api/v1/inventory/product_subgroups", params=filter_params(arguments)))
elif name == "smartkasa_subgroups_create":
return json_response(await http_client.request("POST", "/api/v1/inventory/product_subgroups", data={"subgroup": filter_params(arguments)}))
elif name == "smartkasa_subgroups_update":
sid = arguments.pop("id")
return json_response(await http_client.request("PUT", f"/api/v1/inventory/product_subgroups/{sid}", data={"subgroup": filter_params(arguments)}))
elif name == "smartkasa_subgroups_delete":
return json_response(await http_client.request("DELETE", f"/api/v1/inventory/product_subgroups/{arguments['id']}"))
# ==================== IMPORT ====================
elif name == "smartkasa_import_list":
return json_response(await http_client.request("GET", "/api/v1/inventory/import_products", params=filter_params(arguments)))
elif name == "smartkasa_import_start":
return json_response(await http_client.request("POST", "/api/v1/inventory/import_products", data={"import": filter_params(arguments)}))
elif name == "smartkasa_import_status":
return json_response(await http_client.request("GET", f"/api/v1/inventory/import_products/{arguments['id']}"))
# ==================== SHIFTS ====================
elif name == "smartkasa_shifts_list":
return json_response(await http_client.request("GET", "/api/v1/pos/shifts", params=filter_params(arguments)))
elif name == "smartkasa_shifts_get":
return json_response(await http_client.request("GET", f"/api/v1/pos/shifts/{arguments['id']}"))
# ==================== RECEIPTS ====================
elif name == "smartkasa_receipts_list":
return json_response(await http_client.request("GET", "/api/v1/pos/receipts", params=filter_params(arguments)))
elif name == "smartkasa_receipts_get":
return json_response(await http_client.request("GET", f"/api/v1/pos/receipts/{arguments['id']}"))
elif name == "smartkasa_receipts_create":
return json_response(await http_client.request("POST", "/api/v1/pos/receipts", data={"receipt": filter_params(arguments)}))
elif name == "smartkasa_receipts_update":
rid = arguments.pop("id")
return json_response(await http_client.request("PUT", f"/api/v1/pos/receipts/{rid}", data={"receipt": filter_params(arguments)}))
elif name == "smartkasa_receipts_delete":
return json_response(await http_client.request("DELETE", f"/api/v1/pos/receipts/{arguments['id']}"))
elif name == "smartkasa_receipts_batch_create":
return json_response(await http_client.request("POST", "/api/v1/pos/receipts/batch", data={"receipts": arguments["receipts"]}))
elif name == "smartkasa_receipts_batch_delete":
return json_response(await http_client.request("DELETE", "/api/v1/pos/receipts/batch", data={"batch": {"ids": arguments["ids"]}}))
# ==================== PAYMENT TRANSACTIONS ====================
elif name == "smartkasa_transactions_get":
return json_response(await http_client.request("GET", f"/api/v1/pos/payment_transactions/{arguments['id']}"))
# ==================== REPORTS ====================
elif name == "smartkasa_reports_z_reports":
return json_response(await http_client.request("GET", "/api/v1/reports/z-reports", params=filter_params(arguments)))
elif name == "smartkasa_reports_product_sales":
return json_response(await http_client.request("GET", "/api/v1/stats/product_daily_sales/alter", params=filter_params(arguments)))
else:
logger.warning(f"Unknown tool: {name}")
return json_response({"success": False, "error": f"Unknown tool: {name}"})
except Exception as e:
logger.exception(f"Tool error: {name}")
return json_response({"success": False, "error": str(e)})
# =============================================================================
# MAIN ENTRY POINT
# =============================================================================
def parse_args():
"""Parse command line arguments."""
parser = argparse.ArgumentParser(description="SmartKasa MCP Server")
parser.add_argument(
"--transport",
choices=["stdio", "sse"],
default=config.transport,
help="Transport type: stdio (local) or sse (remote)"
)
parser.add_argument(
"--host",
default=config.sse_host,
help="Host for SSE transport (default: 0.0.0.0)"
)
parser.add_argument(
"--port",
type=int,
default=config.sse_port,
help="Port for SSE transport (default: 8080)"
)
return parser.parse_args()
async def run_stdio():
"""Run server with stdio transport (for local LLM clients)."""
logger.info("Starting SmartKasa MCP Server (stdio transport)")
logger.info(f"Base URL: {config.base_url}")
try:
async with stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
server.create_initialization_options()
)
finally:
await http_client.close()
logger.info("SmartKasa MCP Server stopped")
async def run_sse(host: str, port: int):
"""Run server with SSE transport (for remote deployment)."""
if not SSE_AVAILABLE:
logger.error(
"SSE transport requires additional dependencies: pip install 'mcp[sse]' starlette uvicorn")
sys.exit(1)
logger.info(f"Starting SmartKasa MCP Server (SSE transport)")
logger.info(f"Listening on http://{host}:{port}")
logger.info(f"SSE endpoint: http://{host}:{port}/sse")
logger.info(f"Messages endpoint: http://{host}:{port}/messages")
sse = SseServerTransport("/messages")
async def handle_sse(request):
async with sse.connect_sse(
request.scope, request.receive, request._send
) as streams:
await server.run(
streams[0], streams[1], server.create_initialization_options()
)
async def handle_messages(request):
await sse.handle_post_message(request.scope, request.receive, request._send)
async def health_check(request):
return JSONResponse({
"status": "healthy",
"server": "smartkasa-mcp",
"transport": "sse",
"authenticated": session.is_authenticated()
})
app = Starlette(
debug=config.log_level.upper() == "DEBUG",
routes=[
Route("/sse", endpoint=handle_sse),
Route("/messages", endpoint=handle_messages, methods=["POST"]),
Route("/health", endpoint=health_check),
],
)
uvicorn_config = uvicorn.Config(
app,
host=host,
port=port,
log_level=config.log_level.lower(),
access_log=config.log_level.upper() == "DEBUG"
)
server_instance = uvicorn.Server(uvicorn_config)
try:
await server_instance.serve()
finally:
await http_client.close()
logger.info("SmartKasa MCP Server stopped")
async def main():
"""Run the MCP server with proper lifecycle management."""
args = parse_args()
if args.transport == "sse":
await run_sse(args.host, args.port)
else:
await run_stdio()
if __name__ == "__main__":
asyncio.run(main())
