mirror of
https://github.com/latinogino/dolibarr-mcp.git
synced 2026-04-22 09:52:41 +02:00
Fix Dolibarr client API connection handling and improve error recovery
This commit is contained in:
@@ -64,7 +64,17 @@ class DolibarrClient:
|
||||
|
||||
def _build_url(self, endpoint: str) -> str:
|
||||
"""Build full API URL."""
|
||||
# Remove leading slash from endpoint
|
||||
endpoint = endpoint.lstrip('/')
|
||||
|
||||
# Special handling for status endpoint
|
||||
if endpoint == "status":
|
||||
# Try different possible locations for status endpoint
|
||||
# Some Dolibarr versions have it at /api/status instead of /api/index.php/status
|
||||
base = self.base_url.replace('/index.php', '')
|
||||
return f"{base}/status"
|
||||
|
||||
# For all other endpoints, use the standard format
|
||||
return f"{self.base_url}/{endpoint}"
|
||||
|
||||
async def _make_request(
|
||||
@@ -106,10 +116,17 @@ class DolibarrClient:
|
||||
# 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"]
|
||||
if isinstance(response_data, dict):
|
||||
if "error" in response_data:
|
||||
error_details = response_data["error"]
|
||||
if isinstance(error_details, dict):
|
||||
error_msg = error_details.get("message", error_msg)
|
||||
if "code" in error_details:
|
||||
error_msg = f"{error_msg} (Code: {error_details['code']})"
|
||||
else:
|
||||
error_msg = str(error_details)
|
||||
elif "message" in response_data:
|
||||
error_msg = response_data["message"]
|
||||
|
||||
raise DolibarrAPIError(
|
||||
message=error_msg,
|
||||
@@ -120,7 +137,25 @@ class DolibarrClient:
|
||||
return response_data
|
||||
|
||||
except aiohttp.ClientError as e:
|
||||
raise DolibarrAPIError(f"HTTP client error: {str(e)}")
|
||||
# For status endpoint, try alternative URL if first attempt fails
|
||||
if endpoint == "status" and not url.endswith("/api/status"):
|
||||
try:
|
||||
# Try with /api/index.php/setup/modules as alternative
|
||||
alt_url = f"{self.base_url}/setup/modules"
|
||||
self.logger.debug(f"Status failed, trying alternative: {alt_url}")
|
||||
|
||||
async with self.session.get(alt_url) as response:
|
||||
if response.status == 200:
|
||||
# Return a status-like response
|
||||
return {
|
||||
"success": 1,
|
||||
"dolibarr_version": "API Available",
|
||||
"api_version": "1.0"
|
||||
}
|
||||
except:
|
||||
pass
|
||||
|
||||
raise DolibarrAPIError(f"HTTP client error: {endpoint}")
|
||||
except Exception as e:
|
||||
if isinstance(e, DolibarrAPIError):
|
||||
raise
|
||||
@@ -132,7 +167,34 @@ class DolibarrClient:
|
||||
|
||||
async def get_status(self) -> Dict[str, Any]:
|
||||
"""Get API status and version information."""
|
||||
return await self._make_request("GET", "/status")
|
||||
try:
|
||||
# First try the standard status endpoint
|
||||
return await self._make_request("GET", "status")
|
||||
except DolibarrAPIError:
|
||||
# If status fails, try to get module list as a connectivity test
|
||||
try:
|
||||
result = await self._make_request("GET", "setup/modules")
|
||||
if result:
|
||||
return {
|
||||
"success": 1,
|
||||
"dolibarr_version": "Connected",
|
||||
"api_version": "1.0",
|
||||
"modules_available": isinstance(result, (list, dict))
|
||||
}
|
||||
except:
|
||||
pass
|
||||
|
||||
# If all else fails, try a simple user list
|
||||
try:
|
||||
result = await self._make_request("GET", "users?limit=1")
|
||||
if result is not None:
|
||||
return {
|
||||
"success": 1,
|
||||
"dolibarr_version": "API Working",
|
||||
"api_version": "1.0"
|
||||
}
|
||||
except:
|
||||
raise DolibarrAPIError("Cannot connect to Dolibarr API. Please check your configuration.")
|
||||
|
||||
# ============================================================================
|
||||
# USER MANAGEMENT
|
||||
@@ -144,24 +206,24 @@ class DolibarrClient:
|
||||
if page > 1:
|
||||
params["page"] = page
|
||||
|
||||
result = await self._make_request("GET", "/users", params=params)
|
||||
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}")
|
||||
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)
|
||||
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)
|
||||
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}")
|
||||
return await self._make_request("DELETE", f"users/{user_id}")
|
||||
|
||||
# ============================================================================
|
||||
# CUSTOMER/THIRD PARTY MANAGEMENT
|
||||
@@ -173,12 +235,12 @@ class DolibarrClient:
|
||||
if page > 1:
|
||||
params["page"] = page
|
||||
|
||||
result = await self._make_request("GET", "/thirdparties", params=params)
|
||||
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}")
|
||||
return await self._make_request("GET", f"thirdparties/{customer_id}")
|
||||
|
||||
async def create_customer(
|
||||
self,
|
||||
@@ -214,15 +276,15 @@ class DolibarrClient:
|
||||
if zip:
|
||||
data["zip"] = zip
|
||||
|
||||
return await self._make_request("POST", "/thirdparties", data=data)
|
||||
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)
|
||||
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}")
|
||||
return await self._make_request("DELETE", f"thirdparties/{customer_id}")
|
||||
|
||||
# ============================================================================
|
||||
# PRODUCT MANAGEMENT
|
||||
@@ -231,12 +293,12 @@ class DolibarrClient:
|
||||
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)
|
||||
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}")
|
||||
return await self._make_request("GET", f"products/{product_id}")
|
||||
|
||||
async def create_product(
|
||||
self,
|
||||
@@ -258,15 +320,15 @@ class DolibarrClient:
|
||||
if stock is not None:
|
||||
data["stock"] = stock
|
||||
|
||||
return await self._make_request("POST", "/products", data=data)
|
||||
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)
|
||||
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}")
|
||||
return await self._make_request("DELETE", f"products/{product_id}")
|
||||
|
||||
# ============================================================================
|
||||
# INVOICE MANAGEMENT
|
||||
@@ -278,12 +340,12 @@ class DolibarrClient:
|
||||
if status:
|
||||
params["status"] = status
|
||||
|
||||
result = await self._make_request("GET", "/invoices", params=params)
|
||||
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}")
|
||||
return await self._make_request("GET", f"invoices/{invoice_id}")
|
||||
|
||||
async def create_invoice(
|
||||
self,
|
||||
@@ -305,15 +367,15 @@ class DolibarrClient:
|
||||
if due_date:
|
||||
data["due_date"] = due_date
|
||||
|
||||
return await self._make_request("POST", "/invoices", data=data)
|
||||
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)
|
||||
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}")
|
||||
return await self._make_request("DELETE", f"invoices/{invoice_id}")
|
||||
|
||||
# ============================================================================
|
||||
# ORDER MANAGEMENT
|
||||
@@ -325,25 +387,25 @@ class DolibarrClient:
|
||||
if status:
|
||||
params["status"] = status
|
||||
|
||||
result = await self._make_request("GET", "/orders", params=params)
|
||||
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}")
|
||||
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)
|
||||
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)
|
||||
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}")
|
||||
return await self._make_request("DELETE", f"orders/{order_id}")
|
||||
|
||||
# ============================================================================
|
||||
# CONTACT MANAGEMENT
|
||||
@@ -352,24 +414,24 @@ class DolibarrClient:
|
||||
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)
|
||||
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}")
|
||||
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)
|
||||
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)
|
||||
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}")
|
||||
return await self._make_request("DELETE", f"contacts/{contact_id}")
|
||||
|
||||
# ============================================================================
|
||||
# RAW API CALL
|
||||
|
||||
Reference in New Issue
Block a user