Wavyzz/dolibarr-mcp
1
0
Fork 0
mirror of https://github.com/latinogino/dolibarr-mcp.git synced 2026-04-23 18:15:36 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #15 from latinogino/codex/improve-api-validation-and-error-reporting

Browse Source 
Improve validation, correlation IDs, and client safeguards
This commit is contained in:
latinogino
2026-01-02 17:52:21 +01:00
committed by GitHub
parent 59e91efd30 7356d6c409
commit fb313abd85
9 changed files with 501 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

91
docs/api-reference.md
View File

@@ -69,17 +69,92 @@ empty lists until records are created.
}
```
### Common Error Shape
### Structured Error Shapes
The MCP server and wrapper normalize Dolibarr errors into predictable JSON so
callers can surface actionable messages and debugging breadcrumbs.
#### Validation errors (`400`)
```json
{
"error": {
"code": 404,
"message": "Not found"
}
"error": "Bad Request",
"status": 400,
"message": "Validation failed",
"missing_fields": ["ref", "socid"],
"invalid_fields": [
{ "field": "price", "message": "must be a positive number" }
],
"endpoint": "/api/index.php/products",
"timestamp": "2026-01-02T12:34:56Z"
}
```
Dolibarr communicates detailed failure information in the `error` object. The
client wrapper turns these payloads into Python exceptions with the same
metadata so MCP hosts can display friendly error messages.
#### Unexpected errors (`500`)
```json
{
"error": "Internal Server Error",
"status": 500,
"message": "An unexpected error occurred while creating project",
"correlation_id": "abc123-uuid",
"endpoint": "/api/index.php/projects",
"timestamp": "2026-01-02T12:35:06Z"
}
```
#### Successful create
```json
{
"status": 201,
"id": 42,
"ref": "FREELANCE_HOUR_TEST",
"message": "Product created"
}
```
Include the `correlation_id` from a 500 response when opening support tickets
so logs can be located quickly.
### Create Endpoint Requirements & Examples
The wrapper validates payloads before sending them to Dolibarr. Required fields:
| Endpoint | Required fields | Notes |
| --- | --- | --- |
| `POST /products` | `ref`, `label`, `type`, `price` | `type` accepts `product`/`0` or `service`/`1`. |
| `POST /projects` | `ref`, `name`, `socid` | `title` is accepted as an alias for `name`. |
| `POST /invoices` | `socid` | Provide `lines` for invoice content. |
| `POST /time` (timesheets) | `ref`, `task_id`, `duration`, `fk_project` | Provide `ref` or enable auto-generation. |
#### Example: product create without `ref` (expected 400)
```bash
curl -X POST "https://dolibarr.example/api/index.php/products" \
-H "DOLAPIKEY: <REDACTED>" \
-H "Content-Type: application/json" \
-d '{"label":"Freelance hourly rate test — Gino test","type":"service","price":110.00,"tva_tx":19.0}'
```
#### Example: corrected product payload
```bash
curl -X POST "https://dolibarr.example/api/index.php/products" \
-H "DOLAPIKEY: <REDACTED>" \
-H "Content-Type: application/json" \
-d '{"ref":"FREELANCE_HOUR_TEST","label":"Freelance hourly rate test — Gino test","type":"service","price":110.00,"tva_tx":19.0}'
```
#### Example: project create (minimal payload)
```bash
curl -X POST "https://dolibarr.example/api/index.php/projects" \
-H "DOLAPIKEY: <REDACTED>" \
-H "Content-Type: application/json" \
-d '{"ref":"PERCISION_TEST_PROJECT","name":"Percision test — Project test","socid":8}'
```
If server-side reference auto-generation is enabled, omitting `ref` results in a
predictable `AUTO_<timestamp>` reference. Otherwise, the wrapper will raise a
client-side validation error before sending the request.

8
docs/configuration.md
View File

@@ -9,6 +9,11 @@ host application that will launch the server.
| `DOLIBARR_URL` / `DOLIBARR_SHOP_URL` | Base API URL, e.g. `https://your-dolibarr.example.com/api/index.php` (legacy configs that still export `DOLIBARR_BASE_URL` are also honoured). |
| `DOLIBARR_API_KEY` | Personal Dolibarr API token assigned to your user. |
| `LOG_LEVEL` | Optional logging level (`INFO`, `DEBUG`, `WARNING`, …). |
| `ALLOW_REF_AUTOGEN` | When `true`, the wrapper auto-generates missing `ref` values for create operations. |
| `REF_AUTOGEN_PREFIX` | Prefix used for generated references (default `AUTO`). |
| `DEBUG_MODE` | When `true`, request/response bodies are logged without secrets. |
| `MAX_RETRIES` | Retries for transient HTTP errors (default `2`). |
| `RETRY_BACKOFF_SECONDS` | Base backoff for retries (default `0.5`). |
## Example `.env`
@@ -16,6 +21,9 @@ host application that will launch the server.
DOLIBARR_URL=https://your-dolibarr.example.com/api/index.php
DOLIBARR_API_KEY=your_api_key
LOG_LEVEL=INFO
ALLOW_REF_AUTOGEN=true
REF_AUTOGEN_PREFIX=AUTO
DEBUG_MODE=false
```
The [`Config` class](../src/dolibarr_mcp/config.py) is built with

16
docs/development.md
View File

@@ -59,6 +59,22 @@ The project intentionally avoids heavy linting dependencies. Follow the coding
style already present in the repository and run the test-suite before opening a
pull request.
## Debugging 500 responses with correlation IDs
Unexpected Dolibarr API failures return a `correlation_id` in the JSON body.
Include this value when filing issues or investigating user reports.
1. Capture the `correlation_id` from the HTTP 500 response body.
2. Search the MCP server logs (stdout/stderr) for that ID to locate the full
stack trace. For docker users:
```bash
docker logs <container> 2>&1 | grep "<correlation_id>"
```
3. The log entry contains the failing endpoint and sanitized payload details.
Avoid logging `DOLAPIKEY` or other secrets in bug reports.
## Docker tooling
Container assets live in `docker/`:

25
src/dolibarr_mcp/config.py
View File

@@ -53,6 +53,31 @@ class Config(BaseSettings):
default=8080,
)
allow_ref_autogen: bool = Field(
description="Allow automatic generation of reference fields when missing",
default=False,
)
ref_autogen_prefix: str = Field(
description="Prefix to use when auto-generating references",
default="AUTO",
)
debug_mode: bool = Field(
description="Enable verbose debug logging without exposing secrets",
default=False,
)
max_retries: int = Field(
description="Maximum retries for transient HTTP errors",
default=2,
)
retry_backoff_seconds: float = Field(
description="Base backoff (seconds) for retry strategy",
default=0.5,
)
@field_validator("dolibarr_url")
@classmethod
def validate_dolibarr_url(cls, v: str) -> str:

351
src/dolibarr_mcp/dolibarr_client.py
View File

@@ -1,8 +1,11 @@
"""Professional Dolibarr API client with comprehensive CRUD operations."""
import asyncio
import json
import logging
from typing import Any, Dict, List, Optional
from datetime import datetime
from typing import Any, Dict, List, Optional, Tuple
from uuid import uuid4
import aiohttp
from aiohttp import ClientSession, ClientTimeout
@@ -20,6 +23,10 @@ class DolibarrAPIError(Exception):
super().__init__(self.message)
class DolibarrValidationError(DolibarrAPIError):
"""Raised for client-side validation failures before hitting the API."""
class DolibarrClient:
"""Professional Dolibarr API client with comprehensive functionality."""
@@ -30,9 +37,15 @@ class DolibarrClient:
self.api_key = config.api_key
self.session: Optional[ClientSession] = None
self.logger = logging.getLogger(__name__)
self.debug_mode = getattr(config, "debug_mode", False)
self.allow_ref_autogen = getattr(config, "allow_ref_autogen", False)
self.ref_autogen_prefix = getattr(config, "ref_autogen_prefix", "AUTO")
self.max_retries = getattr(config, "max_retries", 2)
self.retry_backoff_seconds = getattr(config, "retry_backoff_seconds", 0.5)
# Configure timeout
self.timeout = ClientTimeout(total=30, connect=10)
self.logger.setLevel(config.log_level)
async def __aenter__(self):
"""Async context manager entry."""
@@ -104,6 +117,113 @@ class DolibarrClient:
return f"{base}/{endpoint}"
def _mask_api_key(self) -> str:
"""Return a masked representation of the API key for logging."""
if not self.api_key:
return "<not-set>"
if len(self.api_key) <= 6:
return "*" * len(self.api_key)
return f"{self.api_key[:2]}***{self.api_key[-2:]}"
@staticmethod
def _now_iso() -> str:
"""Return current UTC timestamp in ISO format with Z suffix."""
return datetime.utcnow().replace(microsecond=0).isoformat() + "Z"
@staticmethod
def _generate_correlation_id() -> str:
"""Create a unique correlation identifier."""
return str(uuid4())
def _build_validation_error(
self,
endpoint: str,
missing_fields: Optional[List[str]] = None,
invalid_fields: Optional[List[Dict[str, str]]] = None,
message: str = "Validation failed",
status: int = 400,
) -> Dict[str, Any]:
"""Build a structured validation error response."""
return {
"error": "Bad Request",
"status": status,
"message": message,
"missing_fields": missing_fields or [],
"invalid_fields": invalid_fields or [],
"endpoint": f"/{endpoint.lstrip('/')}",
"timestamp": self._now_iso(),
}
def _build_internal_error(self, endpoint: str, message: str, correlation_id: str) -> Dict[str, Any]:
"""Build a structured internal server error response."""
return {
"error": "Internal Server Error",
"status": 500,
"message": message,
"correlation_id": correlation_id,
"endpoint": f"/{endpoint.lstrip('/')}",
"timestamp": self._now_iso(),
}
def _apply_aliases(self, payload: Dict[str, Any], aliases: Dict[str, List[str]]) -> None:
"""Promote alias fields to canonical names."""
for target, options in aliases.items():
if target not in payload:
for alias in options:
if alias in payload and payload[alias] not in (None, ""):
payload[target] = payload.pop(alias)
break
def _validate_payload(
self,
endpoint: str,
payload: Dict[str, Any],
required_fields: List[str],
aliases: Optional[Dict[str, List[str]]] = None,
numeric_positive: Optional[List[str]] = None,
enum_fields: Optional[Dict[str, List[Any]]] = None,
) -> Dict[str, Any]:
"""Validate payload before sending to Dolibarr and optionally auto-generate refs."""
aliases = aliases or {}
numeric_positive = numeric_positive or []
enum_fields = enum_fields or {}
self._apply_aliases(payload, aliases)
missing_fields = [
field
for field in required_fields
if field not in payload or payload[field] in (None, "")
]
invalid_fields: List[Dict[str, str]] = []
for field in numeric_positive:
if field in payload and isinstance(payload[field], (int, float)) and payload[field] < 0:
invalid_fields.append({"field": field, "message": "must be a positive number"})
for field, values in enum_fields.items():
if field in payload and payload[field] not in values:
invalid_fields.append({"field": field, "message": f"must be one of {values}"})
if "ref" in missing_fields and self.allow_ref_autogen:
payload["ref"] = f"{self.ref_autogen_prefix}_{datetime.utcnow().strftime('%Y%m%d%H%M%S')}"
missing_fields = [f for f in missing_fields if f != "ref"]
if missing_fields or invalid_fields:
error_data = self._build_validation_error(
endpoint=endpoint,
missing_fields=missing_fields,
invalid_fields=invalid_fields,
)
raise DolibarrValidationError(
message=error_data["message"],
status_code=error_data["status"],
response_data=error_data,
)
return payload
async def _make_request(
self,
method: str,
@@ -117,76 +237,155 @@ class DolibarrClient:
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):
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,
status_code=response.status,
response_data=response_data
last_exception: Optional[Exception] = None
for attempt in range(self.max_retries + 1):
try:
if self.debug_mode:
self.logger.debug(
"Making %s request to %s with params=%s payload_keys=%s api_key=%s",
method,
url,
params or {},
list((data or {}).keys()),
self._mask_api_key(),
)
return response_data
kwargs = {
"params": params or {},
}
except aiohttp.ClientError as 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}")
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()
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):
# Log response for debugging without leaking secrets
if self.debug_mode:
self.logger.debug("Response status: %s", response.status)
self.logger.debug("Response body (truncated): %s", 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:
if response.status == 400:
missing = []
invalid: List[Dict[str, str]] = []
if isinstance(response_data, dict):
if "missing_fields" in response_data:
missing = response_data.get("missing_fields") or []
if "invalid_fields" in response_data:
invalid = response_data.get("invalid_fields") or []
# Heuristic: derive missing ref from message
if not missing and isinstance(response_data.get("error"), str):
if "ref" in response_data.get("error").lower():
missing.append("ref")
if not missing and "message" in response_data and "ref" in str(response_data["message"]).lower():
missing.append("ref")
error_data = self._build_validation_error(
endpoint=endpoint,
missing_fields=missing,
invalid_fields=invalid,
message="Validation failed",
)
raise DolibarrValidationError(
message=error_data["message"],
status_code=400,
response_data=error_data,
)
if response.status >= 500:
correlation_id = self._generate_correlation_id()
internal_error = self._build_internal_error(
endpoint=endpoint,
message=response_data.get("message", f"An unexpected error occurred while processing {endpoint}"),
correlation_id=correlation_id,
)
self.logger.error(
"Server error %s for %s (correlation_id=%s): %s",
response.status,
endpoint,
correlation_id,
response_text[:500],
)
raise DolibarrAPIError(
message=internal_error["message"],
status_code=response.status,
response_data=internal_error,
)
error_msg = f"HTTP {response.status}: {response.reason}"
if isinstance(response_data, dict):
if "message" in response_data:
error_msg = response_data["message"]
elif "error" in response_data and isinstance(response_data["error"], str):
error_msg = response_data["error"]
raise DolibarrAPIError(
message=error_msg,
status_code=response.status,
response_data=response_data,
)
return response_data
except aiohttp.ClientError as e:
last_exception = e
if endpoint == "status" and not url.endswith("/api/status"):
try:
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 {
"success": 1,
"dolibarr_version": "API Available",
"api_version": "1.0"
}
except Exception as alt_exc: # pylint: disable=broad-except
last_exception = alt_exc
if attempt < self.max_retries and isinstance(e, aiohttp.ClientResponseError) and e.status in {502, 503, 504}:
backoff = self.retry_backoff_seconds * (2 ** attempt)
await asyncio.sleep(backoff)
continue
break
except DolibarrAPIError:
raise
raise DolibarrAPIError(f"Unexpected error: {str(e)}")
except Exception as e: # pylint: disable=broad-except
last_exception = e
break
if isinstance(last_exception, DolibarrAPIError):
raise last_exception
if isinstance(last_exception, Exception):
correlation_id = self._generate_correlation_id()
internal_error = self._build_internal_error(
endpoint=endpoint,
message=str(last_exception),
correlation_id=correlation_id,
)
self.logger.error(
"Unexpected error during %s %s (correlation_id=%s): %s",
method,
endpoint,
correlation_id,
last_exception,
)
raise DolibarrAPIError(
message=internal_error["message"],
status_code=500,
response_data=internal_error,
) from last_exception
raise DolibarrAPIError(f"HTTP client error: {endpoint}")
# ============================================================================
# SYSTEM ENDPOINTS
@@ -359,6 +558,14 @@ class DolibarrClient:
) -> Dict[str, Any]:
"""Create a new product or service."""
payload = self._merge_payload(data, **kwargs)
payload = self._validate_payload(
endpoint="products",
payload=payload,
required_fields=["ref", "label", "type", "price"],
aliases={"label": ["name"]},
numeric_positive=["price"],
enum_fields={"type": ["product", "service", 0, 1]},
)
result = await self.request("POST", "products", data=payload)
return self._extract_identifier(result)
@@ -414,6 +621,12 @@ class DolibarrClient:
if "product_type" in line:
line["product_type"] = line["product_type"]
payload = self._validate_payload(
endpoint="invoices",
payload=payload,
required_fields=["socid"],
)
result = await self.request("POST", "invoices", data=payload)
return self._extract_identifier(result)
@@ -573,6 +786,12 @@ class DolibarrClient:
async def create_project(self, data: Optional[Dict[str, Any]] = None, **kwargs) -> Dict[str, Any]:
"""Create a new project."""
payload = self._merge_payload(data, **kwargs)
payload = self._validate_payload(
endpoint="projects",
payload=payload,
required_fields=["ref", "name", "socid"],
aliases={"name": ["title"]},
)
result = await self.request("POST", "projects", data=payload)
return self._extract_identifier(result)

22
src/dolibarr_mcp/dolibarr_mcp_server.py
View File

@@ -4,6 +4,8 @@ import asyncio
import json
import sys
import logging
import uuid
from datetime import datetime
from contextlib import asynccontextmanager
# Import MCP components
@@ -1377,12 +1379,24 @@ async def handle_call_tool(name: str, arguments: dict):
return [TextContent(type="text", text=json.dumps(result, indent=2))]
except DolibarrAPIError as e:
error_result = {"error": f"Dolibarr API Error: {str(e)}", "type": "api_error"}
return [TextContent(type="text", text=json.dumps(error_result, indent=2))]
error_payload = e.response_data or {
"error": "Dolibarr API Error",
"status": e.status_code or 500,
"message": str(e),
"timestamp": datetime.utcnow().isoformat() + "Z",
}
return [TextContent(type="text", text=json.dumps(error_payload, indent=2))]
except Exception as e:
error_result = {"error": f"Tool execution failed: {str(e)}", "type": "internal_error"}
print(f"🔥 Tool execution error: {e}", file=sys.stderr) # Debug logging
correlation_id = str(uuid.uuid4())
error_result = {
"error": "Internal Server Error",
"status": 500,
"message": f"Tool execution failed: {str(e)}",
"correlation_id": correlation_id,
"timestamp": datetime.utcnow().isoformat() + "Z",
}
print(f"🔥 Tool execution error ({correlation_id}): {e}", file=sys.stderr) # Debug logging
return [TextContent(type="text", text=json.dumps(error_result, indent=2))]

5
tests/test_crud_operations.py
View File

@@ -16,6 +16,7 @@ import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src'))
from dolibarr_mcp import DolibarrClient, Config
from dolibarr_mcp.dolibarr_client import DolibarrValidationError
class TestCRUDOperations:
@@ -78,7 +79,9 @@ class TestCRUDOperations:
# Create
mock_request.return_value = {"id": 10}
product_id = await client.create_product({
"ref": "TEST-PROD",
"label": "Test Product",
"type": "service",
"price": 99.99,
"description": "Test product description"
})
@@ -330,7 +333,7 @@ class TestCRUDOperations:
# Test validation error
mock_request.side_effect = Exception("Validation Error: Missing required field")
with pytest.raises(Exception, match="Validation"):
with pytest.raises(DolibarrValidationError):
await client.create_product({})
# Test connection error

62
tests/test_dolibarr_client.py
View File

@@ -4,7 +4,7 @@ import pytest
from unittest.mock import AsyncMock, patch
from dolibarr_mcp.config import Config
from dolibarr_mcp.dolibarr_client import DolibarrClient, DolibarrAPIError
from dolibarr_mcp.dolibarr_client import DolibarrClient, DolibarrAPIError, DolibarrValidationError
class TestDolibarrClient:
@@ -110,6 +110,66 @@ class TestDolibarrClient:
assert exc_info.value.status_code == 404
assert "Object not found" in str(exc_info.value)
@pytest.mark.asyncio
async def test_validation_error_on_missing_ref(self):
"""Ensure client-side validation catches missing product ref."""
config = Config(
dolibarr_url="https://test.dolibarr.com/api/index.php",
api_key="test_key",
allow_ref_autogen=False,
)
client = DolibarrClient(config)
client.request = AsyncMock(return_value={"id": 1}) # Should not be called
with pytest.raises(DolibarrValidationError) as exc_info:
await client.create_product({"label": "No Ref", "type": "service", "price": 12.5})
assert exc_info.value.response_data["missing_fields"] == ["ref"]
client.request.assert_not_called()
@pytest.mark.asyncio
async def test_autogen_ref_when_enabled(self):
"""Auto-generate refs when allowed by configuration."""
config = Config(
dolibarr_url="https://test.dolibarr.com/api/index.php",
api_key="test_key",
allow_ref_autogen=True,
ref_autogen_prefix="AUTOREF",
)
client = DolibarrClient(config)
client.request = AsyncMock(return_value={"id": 42, "ref": "AUTOREF_123"})
await client.create_product({"label": "Generated Ref Product", "type": "service", "price": 10})
assert client.request.await_count == 1
sent_payload = client.request.call_args.kwargs["data"]
assert "ref" in sent_payload
assert sent_payload["ref"].startswith("AUTOREF_")
@pytest.mark.asyncio
@patch('aiohttp.ClientSession.request')
async def test_internal_error_correlation_id(self, mock_request):
"""Include correlation IDs for unexpected server errors."""
mock_response = AsyncMock()
mock_response.status = 500
mock_response.reason = "Internal Server Error"
mock_response.text.return_value = '{"message": "Database unavailable"}'
mock_request.return_value.__aenter__.return_value = mock_response
config = Config(
dolibarr_url="https://test.dolibarr.com/api/index.php",
api_key="test_key"
)
async with DolibarrClient(config) as client:
with pytest.raises(DolibarrAPIError) as exc_info:
await client.get_project_by_id(1)
assert exc_info.value.status_code == 500
assert "correlation_id" in exc_info.value.response_data
def test_url_building(self):
"""Test URL building functionality."""

1
tests/test_project_operations.py
View File

@@ -42,6 +42,7 @@ class TestProjectOperations:
# Create
mock_request.return_value = {"id": 200}
project_id = await client.create_project({
"ref": "PRJ-NEW-WEBSITE",
"title": "New Website",
"description": "Website redesign project",
"socid": 1,