mirror of
https://github.com/latinogino/dolibarr-mcp.git
synced 2026-04-24 02:25:35 +02:00
feat: Add comprehensive Dolibarr API client with full CRUD operations
This commit is contained in:
latinogino
386
src/dolibarr_mcp/dolibarr_client.py
Normal file
386
src/dolibarr_mcp/dolibarr_client.py
Normal file
@@ -0,0 +1,386 @@
|
||||
"""Professional Dolibarr API client with comprehensive CRUD operations."""
|
||||
|
||||
import json
|
||||
import logging
|
||||
from typing import Any, Dict, List, Optional, Union
|
||||
from urllib.parse import urljoin, quote
|
||||
|
||||
import aiohttp
|
||||
from aiohttp import ClientSession, ClientTimeout
|
||||
|
||||
from .config import Config
|
||||
|
||||
|
||||
class DolibarrAPIError(Exception):
|
||||
"""Custom exception for Dolibarr API errors."""
|
||||
|
||||
def __init__(self, message: str, status_code: Optional[int] = None, response_data: Optional[Dict] = None):
|
||||
self.message = message
|
||||
self.status_code = status_code
|
||||
self.response_data = response_data
|
||||
super().__init__(self.message)
|
||||
|
||||
|
||||
class DolibarrClient:
|
||||
"""Professional Dolibarr API client with comprehensive functionality."""
|
||||
|
||||
def __init__(self, config: Config):
|
||||
"""Initialize the Dolibarr client."""
|
||||
self.config = config
|
||||
self.base_url = config.dolibarr_url.rstrip('/')
|
||||
self.api_key = config.api_key
|
||||
self.session: Optional[ClientSession] = None
|
||||
self.logger = logging.getLogger(__name__)
|
||||
|
||||
# Configure timeout
|
||||
self.timeout = ClientTimeout(total=30, connect=10)
|
||||
|
||||
async def __aenter__(self):
|
||||
"""Async context manager entry."""
|
||||
await self.start_session()
|
||||
return self
|
||||
|
||||
async def __aexit__(self, exc_type, exc_val, exc_tb):
|
||||
"""Async context manager exit."""
|
||||
await self.close_session()
|
||||
|
||||
async def start_session(self):
|
||||
"""Start the HTTP session."""
|
||||
if not self.session:
|
||||
self.session = aiohttp.ClientSession(
|
||||
timeout=self.timeout,
|
||||
headers={
|
||||
"DOLAPIKEY": self.api_key,
|
||||
"Content-Type": "application/json",
|
||||
"Accept": "application/json"
|
||||
}
|
||||
)
|
||||
|
||||
async def close_session(self):
|
||||
"""Close the HTTP session."""
|
||||
if self.session:
|
||||
await self.session.close()
|
||||
self.session = None
|
||||
|
||||
def _build_url(self, endpoint: str) -> str:
|
||||
"""Build full API URL."""
|
||||
endpoint = endpoint.lstrip('/')
|
||||
return f"{self.base_url}/{endpoint}"
|
||||
|
||||
async def _make_request(
|
||||
self,
|
||||
method: str,
|
||||
endpoint: str,
|
||||
params: Optional[Dict] = None,
|
||||
data: Optional[Dict] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""Make HTTP request to Dolibarr API."""
|
||||
if not self.session:
|
||||
await self.start_session()
|
||||
|
||||
url = self._build_url(endpoint)
|
||||
|
||||
try:
|
||||
self.logger.debug(f"Making {method} request to {url}")
|
||||
|
||||
kwargs = {
|
||||
"params": params or {},
|
||||
}
|
||||
|
||||
if data and method.upper() in ["POST", "PUT"]:
|
||||
kwargs["json"] = data
|
||||
|
||||
async with self.session.request(method, url, **kwargs) as response:
|
||||
response_text = await response.text()
|
||||
|
||||
# Log response for debugging
|
||||
self.logger.debug(f"Response status: {response.status}")
|
||||
self.logger.debug(f"Response text: {response_text[:500]}...")
|
||||
|
||||
# Try to parse JSON response
|
||||
try:
|
||||
response_data = json.loads(response_text) if response_text else {}
|
||||
except json.JSONDecodeError:
|
||||
response_data = {"raw_response": response_text}
|
||||
|
||||
# Handle error responses
|
||||
if response.status >= 400:
|
||||
error_msg = f"HTTP {response.status}: {response.reason}"
|
||||
if isinstance(response_data, dict) and "error" in response_data:
|
||||
error_msg = response_data["error"]
|
||||
elif isinstance(response_data, dict) and "message" in response_data:
|
||||
error_msg = response_data["message"]
|
||||
|
||||
raise DolibarrAPIError(
|
||||
message=error_msg,
|
||||
status_code=response.status,
|
||||
response_data=response_data
|
||||
)
|
||||
|
||||
return response_data
|
||||
|
||||
except aiohttp.ClientError as e:
|
||||
raise DolibarrAPIError(f"HTTP client error: {str(e)}")
|
||||
except Exception as e:
|
||||
if isinstance(e, DolibarrAPIError):
|
||||
raise
|
||||
raise DolibarrAPIError(f"Unexpected error: {str(e)}")
|
||||
|
||||
# ============================================================================
|
||||
# SYSTEM ENDPOINTS
|
||||
# ============================================================================
|
||||
|
||||
async def get_status(self) -> Dict[str, Any]:
|
||||
"""Get API status and version information."""
|
||||
return await self._make_request("GET", "/status")
|
||||
|
||||
# ============================================================================
|
||||
# USER MANAGEMENT
|
||||
# ============================================================================
|
||||
|
||||
async def get_users(self, limit: int = 100, page: int = 1) -> List[Dict[str, Any]]:
|
||||
"""Get list of users."""
|
||||
params = {"limit": limit}
|
||||
if page > 1:
|
||||
params["page"] = page
|
||||
|
||||
result = await self._make_request("GET", "/users", params=params)
|
||||
return result if isinstance(result, list) else []
|
||||
|
||||
async def get_user_by_id(self, user_id: int) -> Dict[str, Any]:
|
||||
"""Get specific user by ID."""
|
||||
return await self._make_request("GET", f"/users/{user_id}")
|
||||
|
||||
async def create_user(self, **kwargs) -> Dict[str, Any]:
|
||||
"""Create a new user."""
|
||||
return await self._make_request("POST", "/users", data=kwargs)
|
||||
|
||||
async def update_user(self, user_id: int, **kwargs) -> Dict[str, Any]:
|
||||
"""Update an existing user."""
|
||||
return await self._make_request("PUT", f"/users/{user_id}", data=kwargs)
|
||||
|
||||
async def delete_user(self, user_id: int) -> Dict[str, Any]:
|
||||
"""Delete a user."""
|
||||
return await self._make_request("DELETE", f"/users/{user_id}")
|
||||
|
||||
# ============================================================================
|
||||
# CUSTOMER/THIRD PARTY MANAGEMENT
|
||||
# ============================================================================
|
||||
|
||||
async def get_customers(self, limit: int = 100, page: int = 1) -> List[Dict[str, Any]]:
|
||||
"""Get list of customers/third parties."""
|
||||
params = {"limit": limit}
|
||||
if page > 1:
|
||||
params["page"] = page
|
||||
|
||||
result = await self._make_request("GET", "/thirdparties", params=params)
|
||||
return result if isinstance(result, list) else []
|
||||
|
||||
async def get_customer_by_id(self, customer_id: int) -> Dict[str, Any]:
|
||||
"""Get specific customer by ID."""
|
||||
return await self._make_request("GET", f"/thirdparties/{customer_id}")
|
||||
|
||||
async def create_customer(
|
||||
self,
|
||||
name: str,
|
||||
email: Optional[str] = None,
|
||||
phone: Optional[str] = None,
|
||||
address: Optional[str] = None,
|
||||
town: Optional[str] = None,
|
||||
zip: Optional[str] = None,
|
||||
country_id: int = 1,
|
||||
type: int = 1, # 1=Customer, 2=Supplier, 3=Both
|
||||
status: int = 1,
|
||||
**kwargs
|
||||
) -> Dict[str, Any]:
|
||||
"""Create a new customer/third party."""
|
||||
data = {
|
||||
"name": name,
|
||||
"status": status,
|
||||
"client": type if type in [1, 3] else 0,
|
||||
"fournisseur": 1 if type in [2, 3] else 0,
|
||||
"country_id": country_id,
|
||||
**kwargs
|
||||
}
|
||||
|
||||
if email:
|
||||
data["email"] = email
|
||||
if phone:
|
||||
data["phone"] = phone
|
||||
if address:
|
||||
data["address"] = address
|
||||
if town:
|
||||
data["town"] = town
|
||||
if zip:
|
||||
data["zip"] = zip
|
||||
|
||||
return await self._make_request("POST", "/thirdparties", data=data)
|
||||
|
||||
async def update_customer(self, customer_id: int, **kwargs) -> Dict[str, Any]:
|
||||
"""Update an existing customer."""
|
||||
return await self._make_request("PUT", f"/thirdparties/{customer_id}", data=kwargs)
|
||||
|
||||
async def delete_customer(self, customer_id: int) -> Dict[str, Any]:
|
||||
"""Delete a customer."""
|
||||
return await self._make_request("DELETE", f"/thirdparties/{customer_id}")
|
||||
|
||||
# ============================================================================
|
||||
# PRODUCT MANAGEMENT
|
||||
# ============================================================================
|
||||
|
||||
async def get_products(self, limit: int = 100) -> List[Dict[str, Any]]:
|
||||
"""Get list of products."""
|
||||
params = {"limit": limit}
|
||||
result = await self._make_request("GET", "/products", params=params)
|
||||
return result if isinstance(result, list) else []
|
||||
|
||||
async def get_product_by_id(self, product_id: int) -> Dict[str, Any]:
|
||||
"""Get specific product by ID."""
|
||||
return await self._make_request("GET", f"/products/{product_id}")
|
||||
|
||||
async def create_product(
|
||||
self,
|
||||
label: str,
|
||||
price: float,
|
||||
description: Optional[str] = None,
|
||||
stock: Optional[int] = None,
|
||||
**kwargs
|
||||
) -> Dict[str, Any]:
|
||||
"""Create a new product."""
|
||||
data = {
|
||||
"label": label,
|
||||
"price": price,
|
||||
**kwargs
|
||||
}
|
||||
|
||||
if description:
|
||||
data["description"] = description
|
||||
if stock is not None:
|
||||
data["stock"] = stock
|
||||
|
||||
return await self._make_request("POST", "/products", data=data)
|
||||
|
||||
async def update_product(self, product_id: int, **kwargs) -> Dict[str, Any]:
|
||||
"""Update an existing product."""
|
||||
return await self._make_request("PUT", f"/products/{product_id}", data=kwargs)
|
||||
|
||||
async def delete_product(self, product_id: int) -> Dict[str, Any]:
|
||||
"""Delete a product."""
|
||||
return await self._make_request("DELETE", f"/products/{product_id}")
|
||||
|
||||
# ============================================================================
|
||||
# INVOICE MANAGEMENT
|
||||
# ============================================================================
|
||||
|
||||
async def get_invoices(self, limit: int = 100, status: Optional[str] = None) -> List[Dict[str, Any]]:
|
||||
"""Get list of invoices."""
|
||||
params = {"limit": limit}
|
||||
if status:
|
||||
params["status"] = status
|
||||
|
||||
result = await self._make_request("GET", "/invoices", params=params)
|
||||
return result if isinstance(result, list) else []
|
||||
|
||||
async def get_invoice_by_id(self, invoice_id: int) -> Dict[str, Any]:
|
||||
"""Get specific invoice by ID."""
|
||||
return await self._make_request("GET", f"/invoices/{invoice_id}")
|
||||
|
||||
async def create_invoice(
|
||||
self,
|
||||
customer_id: int,
|
||||
lines: List[Dict[str, Any]],
|
||||
date: Optional[str] = None,
|
||||
due_date: Optional[str] = None,
|
||||
**kwargs
|
||||
) -> Dict[str, Any]:
|
||||
"""Create a new invoice."""
|
||||
data = {
|
||||
"socid": customer_id,
|
||||
"lines": lines,
|
||||
**kwargs
|
||||
}
|
||||
|
||||
if date:
|
||||
data["date"] = date
|
||||
if due_date:
|
||||
data["due_date"] = due_date
|
||||
|
||||
return await self._make_request("POST", "/invoices", data=data)
|
||||
|
||||
async def update_invoice(self, invoice_id: int, **kwargs) -> Dict[str, Any]:
|
||||
"""Update an existing invoice."""
|
||||
return await self._make_request("PUT", f"/invoices/{invoice_id}", data=kwargs)
|
||||
|
||||
async def delete_invoice(self, invoice_id: int) -> Dict[str, Any]:
|
||||
"""Delete an invoice."""
|
||||
return await self._make_request("DELETE", f"/invoices/{invoice_id}")
|
||||
|
||||
# ============================================================================
|
||||
# ORDER MANAGEMENT
|
||||
# ============================================================================
|
||||
|
||||
async def get_orders(self, limit: int = 100, status: Optional[str] = None) -> List[Dict[str, Any]]:
|
||||
"""Get list of orders."""
|
||||
params = {"limit": limit}
|
||||
if status:
|
||||
params["status"] = status
|
||||
|
||||
result = await self._make_request("GET", "/orders", params=params)
|
||||
return result if isinstance(result, list) else []
|
||||
|
||||
async def get_order_by_id(self, order_id: int) -> Dict[str, Any]:
|
||||
"""Get specific order by ID."""
|
||||
return await self._make_request("GET", f"/orders/{order_id}")
|
||||
|
||||
async def create_order(self, customer_id: int, **kwargs) -> Dict[str, Any]:
|
||||
"""Create a new order."""
|
||||
data = {"socid": customer_id, **kwargs}
|
||||
return await self._make_request("POST", "/orders", data=data)
|
||||
|
||||
async def update_order(self, order_id: int, **kwargs) -> Dict[str, Any]:
|
||||
"""Update an existing order."""
|
||||
return await self._make_request("PUT", f"/orders/{order_id}", data=kwargs)
|
||||
|
||||
async def delete_order(self, order_id: int) -> Dict[str, Any]:
|
||||
"""Delete an order."""
|
||||
return await self._make_request("DELETE", f"/orders/{order_id}")
|
||||
|
||||
# ============================================================================
|
||||
# CONTACT MANAGEMENT
|
||||
# ============================================================================
|
||||
|
||||
async def get_contacts(self, limit: int = 100) -> List[Dict[str, Any]]:
|
||||
"""Get list of contacts."""
|
||||
params = {"limit": limit}
|
||||
result = await self._make_request("GET", "/contacts", params=params)
|
||||
return result if isinstance(result, list) else []
|
||||
|
||||
async def get_contact_by_id(self, contact_id: int) -> Dict[str, Any]:
|
||||
"""Get specific contact by ID."""
|
||||
return await self._make_request("GET", f"/contacts/{contact_id}")
|
||||
|
||||
async def create_contact(self, **kwargs) -> Dict[str, Any]:
|
||||
"""Create a new contact."""
|
||||
return await self._make_request("POST", "/contacts", data=kwargs)
|
||||
|
||||
async def update_contact(self, contact_id: int, **kwargs) -> Dict[str, Any]:
|
||||
"""Update an existing contact."""
|
||||
return await self._make_request("PUT", f"/contacts/{contact_id}", data=kwargs)
|
||||
|
||||
async def delete_contact(self, contact_id: int) -> Dict[str, Any]:
|
||||
"""Delete a contact."""
|
||||
return await self._make_request("DELETE", f"/contacts/{contact_id}")
|
||||
|
||||
# ============================================================================
|
||||
# RAW API CALL
|
||||
# ============================================================================
|
||||
|
||||
async def dolibarr_raw_api(
|
||||
self,
|
||||
method: str,
|
||||
endpoint: str,
|
||||
params: Optional[Dict] = None,
|
||||
data: Optional[Dict] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""Make raw API call to any Dolibarr endpoint."""
|
||||
return await self._make_request(method, endpoint, params=params, data=data)
|
||||
Reference in New Issue
Block a user