chaim
e947bca655
Security:
- Path traversal: validate user_id, case_id, skill_name against safe regex
- Input sanitization in memory.py and skills.py file path construction
Correctness:
- ISR timezone: replace hardcoded UTC+3 with ZoneInfo("Asia/Jerusalem") for DST
- save_rule: use EspoCrmApiError.status_code instead of fragile string matching
- Background tasks: store references to prevent GC before completion
- Lock race: use dict.setdefault() for atomic lock creation
- Version: health.py now matches main.py (0.2.0)
- Skill names: normalized to lowercase for dedup consistency
Build:
- Dockerfile: install from pyproject.toml instead of hardcoded pip list
- Remove unused mcp>=1.0.0 dependency from pyproject.toml
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
66 lines
2.0 KiB
Python
66 lines
2.0 KiB
Python
"""EspoCRM REST API client."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import httpx
|
|
import logging
|
|
|
|
logger = logging.getLogger("shira.espocrm")
|
|
|
|
|
|
class EspoCrmApiError(RuntimeError):
|
|
"""Raised when the EspoCRM API returns a non-success status code."""
|
|
|
|
def __init__(self, status_code: int, text: str):
|
|
self.status_code = status_code
|
|
super().__init__(f"EspoCRM API error {status_code}: {text}")
|
|
|
|
|
|
class EspoCrmClient:
|
|
"""Thin async wrapper around the EspoCRM REST API."""
|
|
|
|
def __init__(self, base_url: str, api_key: str, timeout: float = 30.0):
|
|
self.base_url = base_url.rstrip("/")
|
|
self.api_key = api_key
|
|
self.timeout = timeout
|
|
|
|
def _headers(self) -> dict[str, str]:
|
|
return {
|
|
"Content-Type": "application/json",
|
|
"X-Api-Key": self.api_key,
|
|
}
|
|
|
|
async def request(
|
|
self,
|
|
method: str,
|
|
endpoint: str,
|
|
body: dict | None = None,
|
|
) -> dict:
|
|
url = f"{self.base_url}/api/v1/{endpoint}"
|
|
logger.info("[crm-api] %s %s", method, endpoint)
|
|
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
resp = await client.request(
|
|
method,
|
|
url,
|
|
headers=self._headers(),
|
|
json=body,
|
|
)
|
|
if not resp.is_success:
|
|
text = resp.text
|
|
logger.error("[crm-api] Error %s: %s", resp.status_code, text)
|
|
raise EspoCrmApiError(resp.status_code, text)
|
|
return resp.json()
|
|
|
|
async def get(self, endpoint: str) -> dict:
|
|
return await self.request("GET", endpoint)
|
|
|
|
async def post(self, endpoint: str, body: dict) -> dict:
|
|
return await self.request("POST", endpoint, body)
|
|
|
|
async def put(self, endpoint: str, body: dict) -> dict:
|
|
return await self.request("PUT", endpoint, body)
|
|
|
|
async def delete(self, endpoint: str) -> dict:
|
|
return await self.request("DELETE", endpoint)
|