mcp_client.py•2.16 kB
import httpx
from typing import List, Dict, Any, Optional
from app.core.config import MCP_SERVER_HOST, MCP_SERVER_PORT
class MCPClient:
def __init__(self):
self.base_url = f"http://{MCP_SERVER_HOST}:{MCP_SERVER_PORT}/mcp"
async def search_products(self, query: str, category: str = None,
min_price: float = None, max_price: float = None,
limit: int = 5) -> List[Dict[str, Any]]:
params = {
"query": query,
"category": category,
"min_price": min_price,
"max_price": max_price,
"limit": limit
}
async with httpx.AsyncClient() as client:
resp = await client.get(f"{self.base_url}/search_products", params=params)
resp.raise_for_status()
return resp.json()
async def get_product_by_id(self, product_id: str) -> Dict[str, Any]:
async with httpx.AsyncClient() as client:
resp = await client.get(f"{self.base_url}/product/{product_id}")
resp.raise_for_status()
return resp.json()
async def get_products_by_category(self, category: str, limit: int = 10) -> List[Dict[str, Any]]:
params = {"limit": limit}
async with httpx.AsyncClient() as client:
resp = await client.get(f"{self.base_url}/products_by_category/{category}", params=params)
resp.raise_for_status()
return resp.json()
async def count_products(self, category: str = None) -> int:
params = {"category": category} if category else {}
async with httpx.AsyncClient() as client:
resp = await client.get(f"{self.base_url}/count_products", params=params)
resp.raise_for_status()
return resp.json()
async def get_recommendations(self, product_id: str, limit: int = 3) -> List[Dict[str, Any]]:
params = {"limit": limit}
async with httpx.AsyncClient() as client:
resp = await client.get(f"{self.base_url}/recommendations/{product_id}", params=params)
resp.raise_for_status()
return resp.json()