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

Clarify setup guidance and trim dependencies

Browse Source
This commit is contained in:
latinogino
2025-10-12 14:28:23 +02:00
parent a83d70c2b5
commit 81d0d4e89a
47 changed files with 379 additions and 5553 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/dolibarr_mcp/__init__.py
View File

@@ -4,7 +4,7 @@ Dolibarr MCP Server Package
Professional Model Context Protocol server for complete Dolibarr ERP/CRM management.
"""
__version__ = "1.0.0"
__version__ = "1.1.0"
__author__ = "Dolibarr MCP Team"
from .dolibarr_client import DolibarrClient

4
src/dolibarr_mcp/cli.py
View File

@@ -11,7 +11,7 @@ from .dolibarr_mcp_server import main as server_main
@click.group()
@click.version_option(version="1.0.1", prog_name="dolibarr-mcp")
@click.version_option(version="1.1.0", prog_name="dolibarr-mcp")
def cli():
"""Dolibarr MCP Server - Professional ERP integration via Model Context Protocol."""
pass
@@ -91,7 +91,7 @@ def serve(host: str, port: int):
@cli.command()
def version():
"""Show version information."""
click.echo("Dolibarr MCP Server v1.0.1")
click.echo("Dolibarr MCP Server v1.1.0")
click.echo("Professional ERP integration via Model Context Protocol")

159
src/dolibarr_mcp/config.py
View File

@@ -2,10 +2,9 @@
import os
import sys
from typing import Optional
from pydantic import Field, field_validator
from pydantic_settings import BaseSettings
from pydantic import AliasChoices, Field, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
from dotenv import load_dotenv
# Load environment variables from .env file
@@ -14,23 +13,32 @@ load_dotenv()
class Config(BaseSettings):
"""Configuration for Dolibarr MCP Server."""
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
case_sensitive=False,
validate_assignment=True,
extra="forbid",
)
dolibarr_url: str = Field(
description="Dolibarr API URL",
default=""
default="",
)
dolibarr_api_key: str = Field(
description="Dolibarr API key",
default=""
default="",
validation_alias=AliasChoices("dolibarr_api_key", "api_key"),
)
log_level: str = Field(
description="Logging level",
default="INFO"
default="INFO",
)
@field_validator('dolibarr_url')
@field_validator("dolibarr_url")
@classmethod
def validate_dolibarr_url(cls, v: str) -> str:
"""Validate Dolibarr URL."""
@@ -38,62 +46,69 @@ class Config(BaseSettings):
v = os.getenv("DOLIBARR_URL") or os.getenv("DOLIBARR_BASE_URL", "")
if not v:
# Print warning but don't fail
print("⚠️ DOLIBARR_URL/DOLIBARR_BASE_URL not configured - API calls will fail", file=sys.stderr)
print(
"⚠️ DOLIBARR_URL/DOLIBARR_BASE_URL not configured - API calls will fail",
file=sys.stderr,
)
return "https://your-dolibarr-instance.com/api/index.php"
if not v.startswith(('http://', 'https://')):
if not v.startswith(("http://", "https://")):
raise ValueError("DOLIBARR_URL must start with http:// or https://")
# Remove trailing slash if present
v = v.rstrip('/')
v = v.rstrip("/")
# Ensure it ends with the proper API path
if not v.endswith('/api/index.php'):
# Check if it already has /api somewhere
if '/api' in v:
# Just ensure it ends properly
if not v.endswith('/index.php'):
# Check if it ends with /api/index.php/
if v.endswith('/index.php/'):
v = v[:-1] # Remove trailing slash
elif not v.endswith('/index.php'):
v = v + '/index.php'
if not v.endswith("/api/index.php"):
if "/api" in v:
if not v.endswith("/index.php"):
if v.endswith("/index.php/"):
v = v[:-1]
elif not v.endswith("/index.php"):
v = v + "/index.php"
else:
# Add the full API path
v = v + '/api/index.php'
v = v + "/api/index.php"
return v
@field_validator('dolibarr_api_key')
@field_validator("dolibarr_api_key")
@classmethod
def validate_api_key(cls, v: str) -> str:
"""Validate API key."""
if not v:
v = os.getenv("DOLIBARR_API_KEY", "")
if not v:
# Print warning but don't fail
print("⚠️ DOLIBARR_API_KEY not configured - API authentication will fail", file=sys.stderr)
print("📝 Please set DOLIBARR_API_KEY in your .env file or Claude configuration", file=sys.stderr)
print(
"⚠️ DOLIBARR_API_KEY not configured - API authentication will fail",
file=sys.stderr,
)
print(
"📝 Please set DOLIBARR_API_KEY in your .env file or Claude configuration",
file=sys.stderr,
)
return "placeholder_api_key"
if v == "your_dolibarr_api_key_here":
print("⚠️ Using placeholder API key - please configure a real API key", file=sys.stderr)
print(
"⚠️ Using placeholder API key - please configure a real API key",
file=sys.stderr,
)
return v
@field_validator('log_level')
@field_validator("log_level")
@classmethod
def validate_log_level(cls, v: str) -> str:
"""Validate log level."""
if not v:
v = os.getenv("LOG_LEVEL", "INFO")
valid_levels = {'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'}
valid_levels = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"}
if v.upper() not in valid_levels:
print(f"⚠️ Invalid LOG_LEVEL '{v}', using INFO", file=sys.stderr)
return 'INFO'
return "INFO"
return v.upper()
@classmethod
def from_env(cls) -> "Config":
"""Create configuration from environment variables with validation."""
@@ -101,13 +116,15 @@ class Config(BaseSettings):
config = cls(
dolibarr_url=os.getenv("DOLIBARR_URL") or os.getenv("DOLIBARR_BASE_URL", ""),
dolibarr_api_key=os.getenv("DOLIBARR_API_KEY", ""),
log_level=os.getenv("LOG_LEVEL", "INFO")
log_level=os.getenv("LOG_LEVEL", "INFO"),
)
# Debug output for troubleshooting
if os.getenv("DEBUG_CONFIG"):
print(f"✅ Config loaded:", file=sys.stderr)
print(f" URL: {config.dolibarr_url}", file=sys.stderr)
print(f" API Key: {'*' * 10 if config.dolibarr_api_key else 'NOT SET'}", file=sys.stderr)
print(
f" API Key: {'*' * 10 if config.dolibarr_api_key else 'NOT SET'}",
file=sys.stderr,
)
return config
except Exception as e:
print(f"❌ Configuration Error: {e}", file=sys.stderr)
@@ -115,8 +132,14 @@ class Config(BaseSettings):
print("💡 Quick Setup Guide:", file=sys.stderr)
print("1. Copy .env.example to .env", file=sys.stderr)
print("2. Edit .env with your Dolibarr details:", file=sys.stderr)
print(" DOLIBARR_URL=https://your-dolibarr-instance.com", file=sys.stderr)
print(" (or DOLIBARR_BASE_URL=https://your-dolibarr-instance.com/api/index.php/)", file=sys.stderr)
print(
" DOLIBARR_URL=https://your-dolibarr-instance.com",
file=sys.stderr,
)
print(
" (or DOLIBARR_BASE_URL=https://your-dolibarr-instance.com/api/index.php/)",
file=sys.stderr,
)
print(" DOLIBARR_API_KEY=your_api_key_here", file=sys.stderr)
print(file=sys.stderr)
print("🔧 Dolibarr API Key Setup:", file=sys.stderr)
@@ -127,30 +150,22 @@ class Config(BaseSettings):
print(" 5. Create a new API key", file=sys.stderr)
print(file=sys.stderr)
raise
# Alias for backward compatibility
def validate_config(self) -> None:
"""Validate current configuration values."""
self.dolibarr_url = type(self).validate_dolibarr_url(self.dolibarr_url)
self.dolibarr_api_key = type(self).validate_api_key(self.dolibarr_api_key)
self.log_level = type(self).validate_log_level(self.log_level)
if self.dolibarr_url.endswith('your-dolibarr-instance.com/api/index.php') or self.dolibarr_api_key in {'', 'placeholder_api_key', 'your_dolibarr_api_key_here'}:
raise ValueError('Dolibarr configuration is incomplete')
@property
def api_key(self) -> str:
"""Backward compatibility for api_key property."""
return self.dolibarr_api_key
class Config:
"""Pydantic configuration."""
env_file = '.env'
env_file_encoding = 'utf-8'
case_sensitive = False
# Load from environment
env_prefix = ""
@classmethod
def customise_sources(
cls,
init_settings,
env_settings,
file_secret_settings
):
return (
init_settings,
env_settings,
file_secret_settings,
)
@api_key.setter
def api_key(self, value: str) -> None:
"""Allow updating the API key via legacy attribute."""
self.dolibarr_api_key = value

333
src/dolibarr_mcp/dolibarr_client.py
View File

@@ -2,8 +2,7 @@
import json
import logging
from typing import Any, Dict, List, Optional, Union
from urllib.parse import urljoin, quote
from typing import Any, Dict, List, Optional
import aiohttp
from aiohttp import ClientSession, ClientTimeout
@@ -61,22 +60,50 @@ class DolibarrClient:
if self.session:
await self.session.close()
self.session = None
@staticmethod
def _extract_identifier(response: Any) -> Any:
"""Return the identifier from Dolibarr responses when available."""
if isinstance(response, dict):
if "id" in response:
return response["id"]
success = response.get("success")
if isinstance(success, dict) and "id" in success:
return success["id"]
return response
@staticmethod
def _merge_payload(data: Optional[Dict[str, Any]] = None, **kwargs) -> Dict[str, Any]:
"""Merge an optional dictionary with keyword overrides."""
payload: Dict[str, Any] = {}
if data:
payload.update(data)
if kwargs:
payload.update(kwargs)
return payload
async def request(
self,
method: str,
endpoint: str,
params: Optional[Dict] = None,
data: Optional[Dict] = None
) -> Dict[str, Any]:
"""Public helper retained for compatibility with legacy integrations and tests."""
return await self._make_request(method, endpoint, params=params, data=data)
def _build_url(self, endpoint: str) -> str:
"""Build full API URL."""
# Remove leading slash from endpoint
endpoint = endpoint.lstrip('/')
# Special handling for status endpoint
base = self.base_url.rstrip('/')
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}"
base_without_index = base.replace('/index.php', '')
return f"{base_without_index}/status"
return f"{base}/{endpoint}"
async def _make_request(
self,
method: str,
@@ -165,15 +192,19 @@ class DolibarrClient:
# SYSTEM ENDPOINTS
# ============================================================================
async def test_connection(self) -> Dict[str, Any]:
"""Compatibility helper that proxies to get_status."""
return await self.get_status()
async def get_status(self) -> Dict[str, Any]:
"""Get API status and version information."""
try:
# First try the standard status endpoint
return await self._make_request("GET", "status")
return await self.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")
result = await self.request("GET", "setup/modules")
if result:
return {
"success": 1,
@@ -186,7 +217,7 @@ class DolibarrClient:
# If all else fails, try a simple user list
try:
result = await self._make_request("GET", "users?limit=1")
result = await self.request("GET", "users?limit=1")
if result is not None:
return {
"success": 1,
@@ -206,24 +237,36 @@ class DolibarrClient:
if page > 1:
params["page"] = page
result = await self._make_request("GET", "users", params=params)
result = await self.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.request("GET", f"users/{user_id}")
async def create_user(self, **kwargs) -> Dict[str, Any]:
async def create_user(
self,
data: Optional[Dict[str, Any]] = None,
**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]:
payload = self._merge_payload(data, **kwargs)
result = await self.request("POST", "users", data=payload)
return self._extract_identifier(result)
async def update_user(
self,
user_id: int,
data: Optional[Dict[str, Any]] = None,
**kwargs,
) -> Dict[str, Any]:
"""Update an existing user."""
return await self._make_request("PUT", f"users/{user_id}", data=kwargs)
payload = self._merge_payload(data, **kwargs)
return await self.request("PUT", f"users/{user_id}", data=payload)
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.request("DELETE", f"users/{user_id}")
# ============================================================================
# CUSTOMER/THIRD PARTY MANAGEMENT
@@ -235,56 +278,53 @@ class DolibarrClient:
if page > 1:
params["page"] = page
result = await self._make_request("GET", "thirdparties", params=params)
result = await self.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.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
data: Optional[Dict[str, Any]] = None,
**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]:
payload = self._merge_payload(data, **kwargs)
type_value = payload.pop("type", None)
if type_value is not None:
payload.setdefault("client", 1 if type_value in (1, 3) else 0)
payload.setdefault("fournisseur", 1 if type_value in (2, 3) else 0)
else:
payload.setdefault("client", 1)
payload.setdefault("status", payload.get("status", 1))
payload.setdefault("country_id", payload.get("country_id", 1))
result = await self.request("POST", "thirdparties", data=payload)
return self._extract_identifier(result)
async def update_customer(
self,
customer_id: int,
data: Optional[Dict[str, Any]] = None,
**kwargs,
) -> Dict[str, Any]:
"""Update an existing customer."""
return await self._make_request("PUT", f"thirdparties/{customer_id}", data=kwargs)
payload = self._merge_payload(data, **kwargs)
type_value = payload.pop("type", None)
if type_value is not None:
payload["client"] = 1 if type_value in (1, 3) else 0
payload["fournisseur"] = 1 if type_value in (2, 3) else 0
return await self.request("PUT", f"thirdparties/{customer_id}", data=payload)
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.request("DELETE", f"thirdparties/{customer_id}")
# ============================================================================
# PRODUCT MANAGEMENT
@@ -293,60 +333,36 @@ 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.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.request("GET", f"products/{product_id}")
async def create_product(
self,
label: str,
price: float,
ref: Optional[str] = None, # Product reference/SKU
description: Optional[str] = None,
stock: Optional[int] = None,
**kwargs
data: Optional[Dict[str, Any]] = None,
**kwargs,
) -> Dict[str, Any]:
"""Create a new product or service."""
payload = self._merge_payload(data, **kwargs)
result = await self.request("POST", "products", data=payload)
return self._extract_identifier(result)
async def update_product(
self,
product_id: int,
data: Optional[Dict[str, Any]] = None,
**kwargs,
) -> Dict[str, Any]:
"""Create a new product.
Args:
label: Product name/label
price: Product price
ref: Product reference/SKU (required by Dolibarr, auto-generated if not provided)
description: Product description
stock: Initial stock quantity
**kwargs: Additional product fields
"""
import time
# Generate ref if not provided (required field in Dolibarr)
if ref is None:
ref = f"PROD-{int(time.time())}"
data = {
"ref": ref, # Required field
"label": label,
"price": price,
"price_ttc": price, # Price including tax (using same as price for simplicity)
**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)
payload = self._merge_payload(data, **kwargs)
return await self.request("PUT", f"products/{product_id}", data=payload)
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.request("DELETE", f"products/{product_id}")
# ============================================================================
# INVOICE MANAGEMENT
@@ -358,42 +374,36 @@ class DolibarrClient:
if status:
params["status"] = status
result = await self._make_request("GET", "invoices", params=params)
result = await self.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.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
data: Optional[Dict[str, Any]] = 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]:
payload = self._merge_payload(data, **kwargs)
result = await self.request("POST", "invoices", data=payload)
return self._extract_identifier(result)
async def update_invoice(
self,
invoice_id: int,
data: Optional[Dict[str, Any]] = None,
**kwargs,
) -> Dict[str, Any]:
"""Update an existing invoice."""
return await self._make_request("PUT", f"invoices/{invoice_id}", data=kwargs)
payload = self._merge_payload(data, **kwargs)
return await self.request("PUT", f"invoices/{invoice_id}", data=payload)
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.request("DELETE", f"invoices/{invoice_id}")
# ============================================================================
# ORDER MANAGEMENT
@@ -405,25 +415,36 @@ class DolibarrClient:
if status:
params["status"] = status
result = await self._make_request("GET", "orders", params=params)
result = await self.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.request("GET", f"orders/{order_id}")
async def create_order(self, customer_id: int, **kwargs) -> Dict[str, Any]:
async def create_order(
self,
data: Optional[Dict[str, Any]] = None,
**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]:
payload = self._merge_payload(data, **kwargs)
result = await self.request("POST", "orders", data=payload)
return self._extract_identifier(result)
async def update_order(
self,
order_id: int,
data: Optional[Dict[str, Any]] = None,
**kwargs,
) -> Dict[str, Any]:
"""Update an existing order."""
return await self._make_request("PUT", f"orders/{order_id}", data=kwargs)
payload = self._merge_payload(data, **kwargs)
return await self.request("PUT", f"orders/{order_id}", data=payload)
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.request("DELETE", f"orders/{order_id}")
# ============================================================================
# CONTACT MANAGEMENT
@@ -432,24 +453,36 @@ 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.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.request("GET", f"contacts/{contact_id}")
async def create_contact(self, **kwargs) -> Dict[str, Any]:
async def create_contact(
self,
data: Optional[Dict[str, Any]] = None,
**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]:
payload = self._merge_payload(data, **kwargs)
result = await self.request("POST", "contacts", data=payload)
return self._extract_identifier(result)
async def update_contact(
self,
contact_id: int,
data: Optional[Dict[str, Any]] = None,
**kwargs,
) -> Dict[str, Any]:
"""Update an existing contact."""
return await self._make_request("PUT", f"contacts/{contact_id}", data=kwargs)
payload = self._merge_payload(data, **kwargs)
return await self.request("PUT", f"contacts/{contact_id}", data=payload)
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.request("DELETE", f"contacts/{contact_id}")
# ============================================================================
# RAW API CALL
@@ -463,4 +496,4 @@ class DolibarrClient:
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)
return await self.request(method, endpoint, params=params, data=data)

318
src/dolibarr_mcp/simple_client.py
View File

@@ -1,318 +0,0 @@
"""Ultra-simple Dolibarr API client - Windows compatible, zero compiled extensions."""
import json
import logging
import os
import sys
from typing import Any, Dict, List, Optional, Union
# Only use standard library + requests (pure Python)
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
# Minimal config handling without pydantic
class SimpleConfig:
"""Simple configuration without pydantic - no compiled extensions."""
def __init__(self):
# Load .env manually
self.load_env()
self.dolibarr_url = os.getenv("DOLIBARR_URL", "")
self.api_key = os.getenv("DOLIBARR_API_KEY", "")
self.log_level = os.getenv("LOG_LEVEL", "INFO")
# Validate and fix URL
if not self.dolibarr_url or "your-dolibarr-instance" in self.dolibarr_url:
print("⚠️ DOLIBARR_URL not configured in .env file", file=sys.stderr)
self.dolibarr_url = "https://your-dolibarr-instance.com/api/index.php"
if not self.api_key or "your_dolibarr_api_key" in self.api_key:
print("⚠️ DOLIBARR_API_KEY not configured in .env file", file=sys.stderr)
self.api_key = "placeholder_api_key"
# Ensure URL format
if self.dolibarr_url and not self.dolibarr_url.endswith('/api/index.php'):
if '/api' not in self.dolibarr_url:
self.dolibarr_url = self.dolibarr_url.rstrip('/') + '/api/index.php'
elif not self.dolibarr_url.endswith('/index.php'):
self.dolibarr_url = self.dolibarr_url.rstrip('/') + '/index.php'
def load_env(self):
"""Load .env file manually - no python-dotenv needed."""
env_file = '.env'
if os.path.exists(env_file):
with open(env_file, 'r', encoding='utf-8') as f:
for line in f:
line = line.strip()
if line and not line.startswith('#') and '=' in line:
key, value = line.split('=', 1)
os.environ[key.strip()] = value.strip()
class SimpleDolibarrAPIError(Exception):
"""Simple API error exception."""
def __init__(self, message: str, status_code: Optional[int] = None):
self.message = message
self.status_code = status_code
super().__init__(self.message)
class SimpleDolibarrClient:
"""Ultra-simple Dolibarr client using only requests - no aiohttp, no compiled extensions."""
def __init__(self, config: SimpleConfig):
self.config = config
self.base_url = config.dolibarr_url.rstrip('/')
self.api_key = config.api_key
# Create requests session with retries
self.session = requests.Session()
# Add retry strategy
retry_strategy = Retry(
total=3,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504],
)
adapter = HTTPAdapter(max_retries=retry_strategy)
self.session.mount("http://", adapter)
self.session.mount("https://", adapter)
# Set headers
self.session.headers.update({
"DOLAPIKEY": self.api_key,
"Content-Type": "application/json",
"Accept": "application/json",
"User-Agent": "Dolibarr-MCP-Client/1.0"
})
self.logger = logging.getLogger(__name__)
def _build_url(self, endpoint: str) -> str:
"""Build full API URL."""
endpoint = endpoint.lstrip('/')
# Special handling for status endpoint
if endpoint == "status":
base = self.base_url.replace('/index.php', '')
return f"{base}/status"
return f"{self.base_url}/{endpoint}"
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."""
url = self._build_url(endpoint)
try:
self.logger.debug(f"Making {method} request to {url}")
kwargs = {
"params": params or {},
"timeout": 30
}
if data and method.upper() in ["POST", "PUT"]:
kwargs["json"] = data
response = self.session.request(method, url, **kwargs)
# Log response for debugging
self.logger.debug(f"Response status: {response.status_code}")
# Handle error responses
if response.status_code >= 400:
try:
error_data = response.json()
if isinstance(error_data, dict) and "error" in error_data:
error_msg = str(error_data["error"])
else:
error_msg = f"HTTP {response.status_code}: {response.reason}"
except:
error_msg = f"HTTP {response.status_code}: {response.reason}"
raise SimpleDolibarrAPIError(error_msg, response.status_code)
# Try to parse JSON response
try:
return response.json()
except:
# If not JSON, return as text
return {"raw_response": response.text}
except requests.RequestException as e:
# For status endpoint, try alternative
if endpoint == "status":
try:
alt_url = f"{self.base_url.replace('/api/index.php', '')}/setup/modules"
alt_response = self.session.get(alt_url, timeout=10)
if alt_response.status_code == 200:
return {
"success": 1,
"dolibarr_version": "API Available",
"api_version": "1.0"
}
except:
pass
raise SimpleDolibarrAPIError(f"HTTP request failed: {str(e)}")
except Exception as e:
raise SimpleDolibarrAPIError(f"Unexpected error: {str(e)}")
# ============================================================================
# SYSTEM ENDPOINTS
# ============================================================================
def get_status(self) -> Dict[str, Any]:
"""Get API status and version information."""
try:
return self._make_request("GET", "status")
except SimpleDolibarrAPIError:
# Try alternatives
try:
result = self._make_request("GET", "users?limit=1")
if result is not None:
return {
"success": 1,
"dolibarr_version": "API Working",
"api_version": "1.0"
}
except:
pass
raise SimpleDolibarrAPIError("Cannot connect to Dolibarr API. Please check your configuration.")
# ============================================================================
# USER MANAGEMENT
# ============================================================================
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 = self._make_request("GET", "users", params=params)
return result if isinstance(result, list) else []
def get_user_by_id(self, user_id: int) -> Dict[str, Any]:
"""Get specific user by ID."""
return self._make_request("GET", f"users/{user_id}")
def create_user(self, **kwargs) -> Dict[str, Any]:
"""Create a new user."""
return self._make_request("POST", "users", data=kwargs)
def update_user(self, user_id: int, **kwargs) -> Dict[str, Any]:
"""Update an existing user."""
return self._make_request("PUT", f"users/{user_id}", data=kwargs)
def delete_user(self, user_id: int) -> Dict[str, Any]:
"""Delete a user."""
return self._make_request("DELETE", f"users/{user_id}")
# ============================================================================
# CUSTOMER MANAGEMENT
# ============================================================================
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 = self._make_request("GET", "thirdparties", params=params)
return result if isinstance(result, list) else []
def get_customer_by_id(self, customer_id: int) -> Dict[str, Any]:
"""Get specific customer by ID."""
return self._make_request("GET", f"thirdparties/{customer_id}")
def create_customer(self, name: str, **kwargs) -> Dict[str, Any]:
"""Create a new customer."""
data = {
"name": name,
"status": kwargs.get("status", 1),
"client": 1 if kwargs.get("type", 1) in [1, 3] else 0,
"fournisseur": 1 if kwargs.get("type", 1) in [2, 3] else 0,
"country_id": kwargs.get("country_id", 1),
}
# Add optional fields
for field in ["email", "phone", "address", "town", "zip"]:
if field in kwargs:
data[field] = kwargs[field]
return self._make_request("POST", "thirdparties", data=data)
def update_customer(self, customer_id: int, **kwargs) -> Dict[str, Any]:
"""Update an existing customer."""
return self._make_request("PUT", f"thirdparties/{customer_id}", data=kwargs)
def delete_customer(self, customer_id: int) -> Dict[str, Any]:
"""Delete a customer."""
return self._make_request("DELETE", f"thirdparties/{customer_id}")
# ============================================================================
# PRODUCT MANAGEMENT
# ============================================================================
def get_products(self, limit: int = 100) -> List[Dict[str, Any]]:
"""Get list of products."""
params = {"limit": limit}
result = self._make_request("GET", "products", params=params)
return result if isinstance(result, list) else []
def get_product_by_id(self, product_id: int) -> Dict[str, Any]:
"""Get specific product by ID."""
return self._make_request("GET", f"products/{product_id}")
def create_product(self, label: str, price: float, **kwargs) -> Dict[str, Any]:
"""Create a new product."""
import time
# Generate ref if not provided
ref = kwargs.get("ref", f"PROD-{int(time.time())}")
data = {
"ref": ref,
"label": label,
"price": price,
"price_ttc": price,
}
# Add optional fields
for field in ["description", "stock"]:
if field in kwargs:
data[field] = kwargs[field]
return self._make_request("POST", "products", data=data)
def update_product(self, product_id: int, **kwargs) -> Dict[str, Any]:
"""Update an existing product."""
return self._make_request("PUT", f"products/{product_id}", data=kwargs)
def delete_product(self, product_id: int) -> Dict[str, Any]:
"""Delete a product."""
return self._make_request("DELETE", f"products/{product_id}")
# ============================================================================
# RAW API CALL
# ============================================================================
def 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 self._make_request(method, endpoint, params=params, data=data)
def close(self):
"""Close the session."""
if self.session:
self.session.close()

469
src/dolibarr_mcp/standalone_server.py
View File

@@ -1,469 +0,0 @@
"""Standalone Dolibarr MCP Server - Windows Compatible (No pywin32 needed)."""
import asyncio
import json
import sys
import os
import logging
from typing import Any, Dict, List, Optional, Union
# Standard library only - no MCP package needed
from contextlib import asynccontextmanager
# Our Dolibarr components
from .config import Config
from .dolibarr_client import DolibarrClient, DolibarrAPIError
# Configure logging to stderr
logging.basicConfig(
level=logging.WARNING,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
handlers=[logging.StreamHandler(sys.stderr)]
)
class StandaloneMCPServer:
"""Standalone MCP Server implementation without pywin32 dependencies."""
def __init__(self, name: str = "dolibarr-mcp"):
self.name = name
self.logger = logging.getLogger(__name__)
def get_tool_definitions(self) -> List[Dict[str, Any]]:
"""Get all available tool definitions."""
return [
# System & Info
{
"name": "test_connection",
"description": "Test Dolibarr API connection",
"inputSchema": {"type": "object", "properties": {}, "additionalProperties": False}
},
{
"name": "get_status",
"description": "Get Dolibarr system status and version information",
"inputSchema": {"type": "object", "properties": {}, "additionalProperties": False}
},
# User Management CRUD
{
"name": "get_users",
"description": "Get list of users from Dolibarr",
"inputSchema": {
"type": "object",
"properties": {
"limit": {"type": "integer", "description": "Maximum number of users to return (default: 100)", "default": 100},
"page": {"type": "integer", "description": "Page number for pagination (default: 1)", "default": 1}
},
"additionalProperties": False
}
},
{
"name": "get_user_by_id",
"description": "Get specific user details by ID",
"inputSchema": {
"type": "object",
"properties": {
"user_id": {"type": "integer", "description": "User ID to retrieve"}
},
"required": ["user_id"],
"additionalProperties": False
}
},
{
"name": "create_user",
"description": "Create a new user",
"inputSchema": {
"type": "object",
"properties": {
"login": {"type": "string", "description": "User login"},
"lastname": {"type": "string", "description": "Last name"},
"firstname": {"type": "string", "description": "First name"},
"email": {"type": "string", "description": "Email address"},
"password": {"type": "string", "description": "Password"},
"admin": {"type": "integer", "description": "Admin level (0=No, 1=Yes)", "default": 0}
},
"required": ["login", "lastname"],
"additionalProperties": False
}
},
{
"name": "update_user",
"description": "Update an existing user",
"inputSchema": {
"type": "object",
"properties": {
"user_id": {"type": "integer", "description": "User ID to update"},
"login": {"type": "string", "description": "User login"},
"lastname": {"type": "string", "description": "Last name"},
"firstname": {"type": "string", "description": "First name"},
"email": {"type": "string", "description": "Email address"},
"admin": {"type": "integer", "description": "Admin level (0=No, 1=Yes)"}
},
"required": ["user_id"],
"additionalProperties": False
}
},
{
"name": "delete_user",
"description": "Delete a user",
"inputSchema": {
"type": "object",
"properties": {
"user_id": {"type": "integer", "description": "User ID to delete"}
},
"required": ["user_id"],
"additionalProperties": False
}
},
# Customer/Third Party Management CRUD
{
"name": "get_customers",
"description": "Get list of customers/third parties from Dolibarr",
"inputSchema": {
"type": "object",
"properties": {
"limit": {"type": "integer", "description": "Maximum number of customers to return (default: 100)", "default": 100},
"page": {"type": "integer", "description": "Page number for pagination (default: 1)", "default": 1}
},
"additionalProperties": False
}
},
{
"name": "get_customer_by_id",
"description": "Get specific customer details by ID",
"inputSchema": {
"type": "object",
"properties": {
"customer_id": {"type": "integer", "description": "Customer ID to retrieve"}
},
"required": ["customer_id"],
"additionalProperties": False
}
},
{
"name": "create_customer",
"description": "Create a new customer/third party",
"inputSchema": {
"type": "object",
"properties": {
"name": {"type": "string", "description": "Customer name"},
"email": {"type": "string", "description": "Email address"},
"phone": {"type": "string", "description": "Phone number"},
"address": {"type": "string", "description": "Customer address"},
"town": {"type": "string", "description": "City/Town"},
"zip": {"type": "string", "description": "Postal code"},
"country_id": {"type": "integer", "description": "Country ID (default: 1)", "default": 1},
"type": {"type": "integer", "description": "Customer type (1=Customer, 2=Supplier, 3=Both)", "default": 1},
"status": {"type": "integer", "description": "Status (1=Active, 0=Inactive)", "default": 1}
},
"required": ["name"],
"additionalProperties": False
}
},
{
"name": "update_customer",
"description": "Update an existing customer",
"inputSchema": {
"type": "object",
"properties": {
"customer_id": {"type": "integer", "description": "Customer ID to update"},
"name": {"type": "string", "description": "Customer name"},
"email": {"type": "string", "description": "Email address"},
"phone": {"type": "string", "description": "Phone number"},
"address": {"type": "string", "description": "Customer address"},
"town": {"type": "string", "description": "City/Town"},
"zip": {"type": "string", "description": "Postal code"},
"status": {"type": "integer", "description": "Status (1=Active, 0=Inactive)"}
},
"required": ["customer_id"],
"additionalProperties": False
}
},
{
"name": "delete_customer",
"description": "Delete a customer",
"inputSchema": {
"type": "object",
"properties": {
"customer_id": {"type": "integer", "description": "Customer ID to delete"}
},
"required": ["customer_id"],
"additionalProperties": False
}
},
# Product Management CRUD
{
"name": "get_products",
"description": "Get list of products from Dolibarr",
"inputSchema": {
"type": "object",
"properties": {
"limit": {"type": "integer", "description": "Maximum number of products to return (default: 100)", "default": 100}
},
"additionalProperties": False
}
},
{
"name": "get_product_by_id",
"description": "Get specific product details by ID",
"inputSchema": {
"type": "object",
"properties": {
"product_id": {"type": "integer", "description": "Product ID to retrieve"}
},
"required": ["product_id"],
"additionalProperties": False
}
},
{
"name": "create_product",
"description": "Create a new product",
"inputSchema": {
"type": "object",
"properties": {
"label": {"type": "string", "description": "Product name/label"},
"price": {"type": "number", "description": "Product price"},
"description": {"type": "string", "description": "Product description"},
"stock": {"type": "integer", "description": "Initial stock quantity"}
},
"required": ["label", "price"],
"additionalProperties": False
}
},
{
"name": "update_product",
"description": "Update an existing product",
"inputSchema": {
"type": "object",
"properties": {
"product_id": {"type": "integer", "description": "Product ID to update"},
"label": {"type": "string", "description": "Product name/label"},
"price": {"type": "number", "description": "Product price"},
"description": {"type": "string", "description": "Product description"}
},
"required": ["product_id"],
"additionalProperties": False
}
},
{
"name": "delete_product",
"description": "Delete a product",
"inputSchema": {
"type": "object",
"properties": {
"product_id": {"type": "integer", "description": "Product ID to delete"}
},
"required": ["product_id"],
"additionalProperties": False
}
},
# Raw API Access
{
"name": "dolibarr_raw_api",
"description": "Make raw API call to any Dolibarr endpoint",
"inputSchema": {
"type": "object",
"properties": {
"method": {"type": "string", "description": "HTTP method", "enum": ["GET", "POST", "PUT", "DELETE"]},
"endpoint": {"type": "string", "description": "API endpoint (e.g., /thirdparties, /invoices)"},
"params": {"type": "object", "description": "Query parameters"},
"data": {"type": "object", "description": "Request payload for POST/PUT requests"}
},
"required": ["method", "endpoint"],
"additionalProperties": False
}
}
]
async def handle_tool_call(self, name: str, arguments: dict) -> Dict[str, Any]:
"""Handle tool calls using the DolibarrClient."""
try:
# Initialize the config and client
config = Config()
async with DolibarrClient(config) as client:
# System & Info
if name == "test_connection":
result = await client.get_status()
if 'success' not in result:
result = {"status": "success", "message": "API connection working", "data": result}
elif name == "get_status":
result = await client.get_status()
# User Management
elif name == "get_users":
result = await client.get_users(
limit=arguments.get('limit', 100),
page=arguments.get('page', 1)
)
elif name == "get_user_by_id":
result = await client.get_user_by_id(arguments['user_id'])
elif name == "create_user":
result = await client.create_user(**arguments)
elif name == "update_user":
user_id = arguments.pop('user_id')
result = await client.update_user(user_id, **arguments)
elif name == "delete_user":
result = await client.delete_user(arguments['user_id'])
# Customer Management
elif name == "get_customers":
result = await client.get_customers(
limit=arguments.get('limit', 100),
page=arguments.get('page', 1)
)
elif name == "get_customer_by_id":
result = await client.get_customer_by_id(arguments['customer_id'])
elif name == "create_customer":
result = await client.create_customer(**arguments)
elif name == "update_customer":
customer_id = arguments.pop('customer_id')
result = await client.update_customer(customer_id, **arguments)
elif name == "delete_customer":
result = await client.delete_customer(arguments['customer_id'])
# Product Management
elif name == "get_products":
result = await client.get_products(limit=arguments.get('limit', 100))
elif name == "get_product_by_id":
result = await client.get_product_by_id(arguments['product_id'])
elif name == "create_product":
result = await client.create_product(**arguments)
elif name == "update_product":
product_id = arguments.pop('product_id')
result = await client.update_product(product_id, **arguments)
elif name == "delete_product":
result = await client.delete_product(arguments['product_id'])
# Raw API Access
elif name == "dolibarr_raw_api":
result = await client.dolibarr_raw_api(**arguments)
else:
result = {"error": f"Unknown tool: {name}"}
return {"success": True, "data": result}
except DolibarrAPIError as e:
return {"error": f"Dolibarr API Error: {str(e)}", "type": "api_error"}
except Exception as e:
self.logger.error(f"Tool execution error: {e}")
return {"error": f"Tool execution failed: {str(e)}", "type": "internal_error"}
def format_response(self, content: Dict[str, Any]) -> str:
"""Format response as JSON string."""
return json.dumps(content, indent=2, ensure_ascii=False)
async def run_interactive(self):
"""Run server in interactive mode for testing."""
print("🚀 Standalone Dolibarr MCP Server (Windows Compatible)", file=sys.stderr)
print("✅ NO pywin32 dependencies required!", file=sys.stderr)
print("", file=sys.stderr)
# Test API connection
try:
config = Config()
if not config.dolibarr_url or config.dolibarr_url.startswith("https://your-dolibarr-instance"):
print("⚠️ DOLIBARR_URL not configured in .env file", file=sys.stderr)
print("📝 Please edit .env with your Dolibarr credentials", file=sys.stderr)
elif not config.api_key or config.api_key in ["your_dolibarr_api_key_here", "placeholder_api_key"]:
print("⚠️ DOLIBARR_API_KEY not configured in .env file", file=sys.stderr)
print("📝 Please edit .env with your Dolibarr API key", file=sys.stderr)
else:
print("🧪 Testing Dolibarr API connection...", file=sys.stderr)
test_result = await self.handle_tool_call("test_connection", {})
if test_result.get("success"):
print("✅ Dolibarr API connection successful!", file=sys.stderr)
else:
print(f"⚠️ API test result: {test_result}", file=sys.stderr)
except Exception as e:
print(f"⚠️ Configuration error: {e}", file=sys.stderr)
print("", file=sys.stderr)
print("📋 Available Tools:", file=sys.stderr)
tools = self.get_tool_definitions()
for tool in tools:
print(f" • {tool['name']} - {tool['description']}", file=sys.stderr)
print("", file=sys.stderr)
print("💡 Interactive Testing Mode:", file=sys.stderr)
print(" Type 'list' to see all tools", file=sys.stderr)
print(" Type 'test <tool_name>' to test a tool", file=sys.stderr)
print(" Type 'exit' to quit", file=sys.stderr)
print("", file=sys.stderr)
while True:
try:
command = input("dolibarr-mcp> ").strip()
if command == "exit":
break
elif command == "list":
print("Available tools:")
for tool in tools:
print(f" {tool['name']} - {tool['description']}")
elif command.startswith("test "):
tool_name = command[5:].strip()
if tool_name == "test_connection":
result = await self.handle_tool_call("test_connection", {})
print(self.format_response(result))
elif tool_name == "get_status":
result = await self.handle_tool_call("get_status", {})
print(self.format_response(result))
elif tool_name == "get_users":
result = await self.handle_tool_call("get_users", {"limit": 5})
print(self.format_response(result))
elif tool_name == "get_customers":
result = await self.handle_tool_call("get_customers", {"limit": 5})
print(self.format_response(result))
elif tool_name == "get_products":
result = await self.handle_tool_call("get_products", {"limit": 5})
print(self.format_response(result))
else:
print(f"Tool '{tool_name}' requires parameters. Available quick tests: test_connection, get_status, get_users, get_customers, get_products")
elif command:
print("Unknown command. Use 'list', 'test <tool_name>', or 'exit'")
except KeyboardInterrupt:
break
except Exception as e:
print(f"Error: {e}")
print("\n👋 Goodbye!")
async def main():
"""Main entry point."""
server = StandaloneMCPServer("dolibarr-mcp-standalone")
await server.run_interactive()
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
print("\n👋 Server stopped by user", file=sys.stderr)
sys.exit(0)
except Exception as e:
print(f"❌ Server error: {e}", file=sys.stderr)
sys.exit(1)

607
src/dolibarr_mcp/ultra_simple_server.py
View File

@@ -1,607 +0,0 @@
"""Ultra-simple Dolibarr server - COMPLETELY SELF-CONTAINED - Zero external dependencies issues."""
import json
import sys
import os
import logging
from typing import Any, Dict, List, Optional
# EVERYTHING is self-contained in this single file to avoid import issues
# ============================================================================
# SELF-CONTAINED CONFIGURATION CLASS
# ============================================================================
class UltraSimpleConfig:
"""Ultra-simple configuration - completely self-contained."""
def __init__(self):
# Load .env manually
self.load_env()
self.dolibarr_url = os.getenv("DOLIBARR_URL", "")
self.api_key = os.getenv("DOLIBARR_API_KEY", "")
self.log_level = os.getenv("LOG_LEVEL", "INFO")
# Validate and fix URL
if not self.dolibarr_url or "your-dolibarr-instance" in self.dolibarr_url:
print("⚠️ DOLIBARR_URL not configured in .env file", file=sys.stderr)
self.dolibarr_url = "https://your-dolibarr-instance.com/api/index.php"
if not self.api_key or "your_dolibarr_api_key" in self.api_key:
print("⚠️ DOLIBARR_API_KEY not configured in .env file", file=sys.stderr)
self.api_key = "placeholder_api_key"
# Normalize URL - remove trailing slashes and ensure proper format
self.dolibarr_url = self.dolibarr_url.rstrip('/')
# If URL doesn't contain /api/index.php, try to add it
if '/api/index.php' not in self.dolibarr_url:
if '/api' in self.dolibarr_url:
# Has /api but not /index.php
if not self.dolibarr_url.endswith('/index.php'):
self.dolibarr_url = self.dolibarr_url + '/index.php'
else:
# No /api at all - add full path
self.dolibarr_url = self.dolibarr_url + '/api/index.php'
# Debug output
print(f"🔧 Configuration loaded:", file=sys.stderr)
print(f" URL: {self.dolibarr_url}", file=sys.stderr)
print(f" API Key: {'*' * min(len(self.api_key), 10)}... (length: {len(self.api_key)})", file=sys.stderr)
def load_env(self):
"""Load .env file manually - no python-dotenv needed."""
env_file = '.env'
if os.path.exists(env_file):
print(f"📄 Loading environment from {env_file}", file=sys.stderr)
with open(env_file, 'r', encoding='utf-8') as f:
for line in f:
line = line.strip()
if line and not line.startswith('#') and '=' in line:
key, value = line.split('=', 1)
key = key.strip()
value = value.strip()
# Remove quotes if present
if value.startswith('"') and value.endswith('"'):
value = value[1:-1]
elif value.startswith("'") and value.endswith("'"):
value = value[1:-1]
os.environ[key] = value
print(f" Loaded: {key} = {value[:30]}..." if len(value) > 30 else f" Loaded: {key}", file=sys.stderr)
else:
print(f"⚠️ No .env file found in current directory", file=sys.stderr)
# ============================================================================
# SELF-CONTAINED API CLIENT
# ============================================================================
class UltraSimpleAPIError(Exception):
"""Simple API error exception."""
def __init__(self, message: str, status_code: Optional[int] = None):
self.message = message
self.status_code = status_code
super().__init__(self.message)
class UltraSimpleAPIClient:
"""Ultra-simple Dolibarr client - completely self-contained."""
def __init__(self, config: UltraSimpleConfig):
self.config = config
self.base_url = config.dolibarr_url.rstrip('/')
self.api_key = config.api_key
self.logger = logging.getLogger(__name__)
# We'll use requests when needed, imported inside methods
def _build_url(self, endpoint: str) -> str:
"""Build full API URL."""
endpoint = endpoint.lstrip('/')
# For status endpoint, try different variations
if endpoint == "status":
# First try the standard status endpoint
return f"{self.base_url}/status"
# For other endpoints, just append to base URL
return f"{self.base_url}/{endpoint}"
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."""
# Import requests here to avoid import issues
try:
import requests
except ImportError:
raise UltraSimpleAPIError("requests library not available. Please run setup_ultra.bat")
url = self._build_url(endpoint)
try:
self.logger.debug(f"Making {method} request to {url}")
print(f"🔍 API Request: {method} {url}", file=sys.stderr)
headers = {
"DOLAPIKEY": self.api_key,
"Content-Type": "application/json",
"Accept": "application/json",
"User-Agent": "Dolibarr-MCP-Ultra/1.0"
}
# Debug headers (without full API key)
print(f" Headers: DOLAPIKEY={self.api_key[:10]}...", file=sys.stderr)
kwargs = {
"params": params or {},
"timeout": 30,
"headers": headers,
"verify": True # Enable SSL verification
}
if data and method.upper() in ["POST", "PUT"]:
kwargs["json"] = data
response = requests.request(method, url, **kwargs)
print(f" Response Status: {response.status_code}", file=sys.stderr)
# Handle error responses
if response.status_code >= 400:
print(f" Response Content: {response.text[:500]}", file=sys.stderr)
try:
error_data = response.json()
if isinstance(error_data, dict):
if "error" in error_data:
error_msg = error_data["error"].get("message", str(error_data["error"]))
elif "errors" in error_data:
error_msg = str(error_data["errors"])
else:
error_msg = f"HTTP {response.status_code}: {response.reason}"
else:
error_msg = f"HTTP {response.status_code}: {response.text[:200]}"
except:
error_msg = f"HTTP {response.status_code}: {response.reason}"
raise UltraSimpleAPIError(error_msg, response.status_code)
# Try to parse JSON response
try:
result = response.json()
print(f" ✅ Response OK: {type(result)}", file=sys.stderr)
return result
except:
print(f" ⚠️ Non-JSON response: {response.text[:100]}", file=sys.stderr)
return {"raw_response": response.text}
except requests.RequestException as e:
print(f" ❌ Request failed: {str(e)}", file=sys.stderr)
# For connection errors, provide more helpful messages
if "SSLError" in str(e.__class__.__name__):
raise UltraSimpleAPIError(f"SSL Error: {str(e)}. Try checking if the URL is correct and the SSL certificate is valid.")
elif "ConnectionError" in str(e.__class__.__name__):
raise UltraSimpleAPIError(f"Connection Error: Cannot reach {url}. Please check your URL and network connection.")
elif "Timeout" in str(e.__class__.__name__):
raise UltraSimpleAPIError(f"Timeout: The server took too long to respond. Please check if the URL is correct.")
raise UltraSimpleAPIError(f"HTTP request failed: {str(e)}")
except Exception as e:
print(f" ❌ Unexpected error: {str(e)}", file=sys.stderr)
raise UltraSimpleAPIError(f"Unexpected error: {str(e)}")
# API Methods
def get_status(self) -> Dict[str, Any]:
"""Get API status - try multiple approaches."""
# First try the login endpoint which is commonly available
try:
print("🔍 Attempting to verify API access via login endpoint...", file=sys.stderr)
login_data = {
"login": "test",
"password": "test",
"reset": 0
}
# Don't actually login, just check if the endpoint responds
self._make_request("POST", "login", data=login_data)
except UltraSimpleAPIError as e:
# If we get a 403 or 401, it means the API is working but credentials are wrong
if e.status_code in [401, 403]:
print(" ✅ API is reachable (authentication endpoint responded)", file=sys.stderr)
return {
"success": 1,
"dolibarr_version": "API Working",
"api_version": "1.0",
"message": "API is reachable and responding"
}
except:
pass
# Try to get users as a status check
try:
print("🔍 Attempting to verify API access via users endpoint...", file=sys.stderr)
result = self._make_request("GET", "users", params={"limit": 1})
if result is not None:
return {
"success": 1,
"dolibarr_version": "API Working",
"api_version": "1.0",
"users_accessible": True
}
except:
pass
# Try the status endpoint
try:
print("🔍 Attempting standard status endpoint...", file=sys.stderr)
return self._make_request("GET", "status")
except:
pass
# Last resort - try to get any response
try:
print("🔍 Testing basic API connectivity...", file=sys.stderr)
# Try a simple GET to the base API URL
import requests
response = requests.get(
self.base_url,
headers={"DOLAPIKEY": self.api_key},
timeout=10,
verify=True
)
if response.status_code < 500:
return {
"success": 1,
"dolibarr_version": "API Endpoint Exists",
"api_version": "Unknown",
"status_code": response.status_code
}
except:
pass
raise UltraSimpleAPIError("Cannot connect to Dolibarr API. Please check your configuration.")
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 = self._make_request("GET", "users", params=params)
return result if isinstance(result, list) else []
def get_user_by_id(self, user_id: int) -> Dict[str, Any]:
"""Get specific user by ID."""
return self._make_request("GET", f"users/{user_id}")
def create_user(self, **kwargs) -> Dict[str, Any]:
"""Create a new user."""
return self._make_request("POST", "users", data=kwargs)
def update_user(self, user_id: int, **kwargs) -> Dict[str, Any]:
"""Update an existing user."""
return self._make_request("PUT", f"users/{user_id}", data=kwargs)
def delete_user(self, user_id: int) -> Dict[str, Any]:
"""Delete a user."""
return self._make_request("DELETE", f"users/{user_id}")
def get_customers(self, limit: int = 100, page: int = 1) -> List[Dict[str, Any]]:
"""Get list of customers."""
params = {"limit": limit}
if page > 1:
params["page"] = page
result = self._make_request("GET", "thirdparties", params=params)
return result if isinstance(result, list) else []
def get_customer_by_id(self, customer_id: int) -> Dict[str, Any]:
"""Get specific customer by ID."""
return self._make_request("GET", f"thirdparties/{customer_id}")
def create_customer(self, name: str, **kwargs) -> Dict[str, Any]:
"""Create a new customer."""
data = {
"name": name,
"status": kwargs.get("status", 1),
"client": 1 if kwargs.get("type", 1) in [1, 3] else 0,
"fournisseur": 1 if kwargs.get("type", 1) in [2, 3] else 0,
"country_id": kwargs.get("country_id", 1),
}
for field in ["email", "phone", "address", "town", "zip"]:
if field in kwargs:
data[field] = kwargs[field]
return self._make_request("POST", "thirdparties", data=data)
def update_customer(self, customer_id: int, **kwargs) -> Dict[str, Any]:
"""Update an existing customer."""
return self._make_request("PUT", f"thirdparties/{customer_id}", data=kwargs)
def delete_customer(self, customer_id: int) -> Dict[str, Any]:
"""Delete a customer."""
return self._make_request("DELETE", f"thirdparties/{customer_id}")
def get_products(self, limit: int = 100) -> List[Dict[str, Any]]:
"""Get list of products."""
params = {"limit": limit}
result = self._make_request("GET", "products", params=params)
return result if isinstance(result, list) else []
def get_product_by_id(self, product_id: int) -> Dict[str, Any]:
"""Get specific product by ID."""
return self._make_request("GET", f"products/{product_id}")
def create_product(self, label: str, price: float, **kwargs) -> Dict[str, Any]:
"""Create a new product."""
import time
ref = kwargs.get("ref", f"PROD-{int(time.time())}")
data = {
"ref": ref,
"label": label,
"price": price,
"price_ttc": price,
}
for field in ["description", "stock"]:
if field in kwargs:
data[field] = kwargs[field]
return self._make_request("POST", "products", data=data)
def update_product(self, product_id: int, **kwargs) -> Dict[str, Any]:
"""Update an existing product."""
return self._make_request("PUT", f"products/{product_id}", data=kwargs)
def delete_product(self, product_id: int) -> Dict[str, Any]:
"""Delete a product."""
return self._make_request("DELETE", f"products/{product_id}")
def raw_api(self, method: str, endpoint: str, params: Optional[Dict] = None, data: Optional[Dict] = None) -> Dict[str, Any]:
"""Make raw API call."""
return self._make_request(method, endpoint, params=params, data=data)
# ============================================================================
# ULTRA-SIMPLE SERVER
# ============================================================================
class UltraSimpleServer:
"""Ultra-simple server - completely self-contained."""
def __init__(self, name: str = "dolibarr-mcp-ultra"):
self.name = name
self.logger = logging.getLogger(__name__)
self.client = None
def init_client(self):
"""Initialize the Dolibarr client."""
if not self.client:
config = UltraSimpleConfig()
self.client = UltraSimpleAPIClient(config)
def get_available_tools(self) -> List[str]:
"""Get list of available tool names."""
return [
"test_connection", "get_status", "get_users", "get_user_by_id",
"create_user", "update_user", "delete_user", "get_customers",
"get_customer_by_id", "create_customer", "update_customer",
"delete_customer", "get_products", "get_product_by_id",
"create_product", "update_product", "delete_product", "raw_api"
]
def handle_tool_call(self, tool_name: str, arguments: dict) -> Dict[str, Any]:
"""Handle tool calls."""
try:
self.init_client()
if tool_name == "test_connection":
result = self.client.get_status()
if 'success' not in result:
result = {"status": "success", "message": "API connection working", "data": result}
return {"success": True, "data": result}
elif tool_name == "get_status":
result = self.client.get_status()
return {"success": True, "data": result}
elif tool_name == "get_users":
result = self.client.get_users(
limit=arguments.get('limit', 100),
page=arguments.get('page', 1)
)
return {"success": True, "data": result}
elif tool_name == "get_user_by_id":
result = self.client.get_user_by_id(arguments['user_id'])
return {"success": True, "data": result}
elif tool_name == "create_user":
result = self.client.create_user(**arguments)
return {"success": True, "data": result}
elif tool_name == "update_user":
user_id = arguments.pop('user_id')
result = self.client.update_user(user_id, **arguments)
return {"success": True, "data": result}
elif tool_name == "delete_user":
result = self.client.delete_user(arguments['user_id'])
return {"success": True, "data": result}
elif tool_name == "get_customers":
result = self.client.get_customers(
limit=arguments.get('limit', 100),
page=arguments.get('page', 1)
)
return {"success": True, "data": result}
elif tool_name == "get_customer_by_id":
result = self.client.get_customer_by_id(arguments['customer_id'])
return {"success": True, "data": result}
elif tool_name == "create_customer":
result = self.client.create_customer(**arguments)
return {"success": True, "data": result}
elif tool_name == "update_customer":
customer_id = arguments.pop('customer_id')
result = self.client.update_customer(customer_id, **arguments)
return {"success": True, "data": result}
elif tool_name == "delete_customer":
result = self.client.delete_customer(arguments['customer_id'])
return {"success": True, "data": result}
elif tool_name == "get_products":
result = self.client.get_products(limit=arguments.get('limit', 100))
return {"success": True, "data": result}
elif tool_name == "get_product_by_id":
result = self.client.get_product_by_id(arguments['product_id'])
return {"success": True, "data": result}
elif tool_name == "create_product":
result = self.client.create_product(**arguments)
return {"success": True, "data": result}
elif tool_name == "update_product":
product_id = arguments.pop('product_id')
result = self.client.update_product(product_id, **arguments)
return {"success": True, "data": result}
elif tool_name == "delete_product":
result = self.client.delete_product(arguments['product_id'])
return {"success": True, "data": result}
elif tool_name == "raw_api":
result = self.client.raw_api(**arguments)
return {"success": True, "data": result}
else:
return {"error": f"Unknown tool: {tool_name}", "type": "unknown_tool"}
except UltraSimpleAPIError as e:
return {"error": f"Dolibarr API Error: {str(e)}", "type": "api_error"}
except Exception as e:
self.logger.error(f"Tool execution error: {e}")
return {"error": f"Tool execution failed: {str(e)}", "type": "internal_error"}
def format_response(self, content: Dict[str, Any]) -> str:
"""Format response as JSON string."""
return json.dumps(content, indent=2, ensure_ascii=False)
def run_interactive(self):
"""Run server in interactive mode."""
print("=" * 70, file=sys.stderr)
print("Dolibarr MCP ULTRA Server", file=sys.stderr)
print("=" * 70, file=sys.stderr)
print("Maximum Windows Compatibility Mode", file=sys.stderr)
print("ZERO compiled extensions (.pyd files)", file=sys.stderr)
print("Activating ultra virtual environment...", file=sys.stderr)
print("🚀 Starting ULTRA-COMPATIBLE Dolibarr MCP Server...", file=sys.stderr)
print("├─ Pure Python implementation", file=sys.stderr)
print("├─ ZERO compiled extensions", file=sys.stderr)
print("├─ Standard library + requests only", file=sys.stderr)
print("└─ Works on ANY Windows version", file=sys.stderr)
print("", file=sys.stderr)
print("Available features:", file=sys.stderr)
print(" • All CRUD operations for Dolibarr", file=sys.stderr)
print(" • Interactive testing console", file=sys.stderr)
print(" • Professional error handling", file=sys.stderr)
print(" • Zero permission issues", file=sys.stderr)
print("", file=sys.stderr)
print("🚀 Ultra-Simple Dolibarr MCP Server (Maximum Windows Compatibility)", file=sys.stderr)
print("✅ ZERO compiled extensions - NO .pyd files!", file=sys.stderr)
print("✅ Completely self-contained - no import issues!", file=sys.stderr)
print("", file=sys.stderr)
# Test configuration
try:
config = UltraSimpleConfig()
if "your-dolibarr-instance" in config.dolibarr_url:
print("⚠️ DOLIBARR_URL not configured in .env file", file=sys.stderr)
print("📝 Please edit .env with your Dolibarr credentials", file=sys.stderr)
elif "placeholder_api_key" in config.api_key:
print("⚠️ DOLIBARR_API_KEY not configured in .env file", file=sys.stderr)
print("📝 Please edit .env with your Dolibarr API key", file=sys.stderr)
else:
print("🧪 Testing Dolibarr API connection...", file=sys.stderr)
test_result = self.handle_tool_call("test_connection", {})
if test_result.get("success"):
print("✅ Dolibarr API connection successful!", file=sys.stderr)
else:
print(f"⚠️ API test result: {test_result}", file=sys.stderr)
except Exception as e:
print(f"⚠️ Configuration error: {e}", file=sys.stderr)
print("", file=sys.stderr)
print("📋 Available Tools:", file=sys.stderr)
tools = self.get_available_tools()
for i, tool in enumerate(tools, 1):
print(f" {i:2}. {tool}", file=sys.stderr)
print("", file=sys.stderr)
print("💡 Interactive Testing Mode:", file=sys.stderr)
print(" Type 'list' to see all tools", file=sys.stderr)
print(" Type 'test <tool_name>' to test a tool", file=sys.stderr)
print(" Type 'help' for more commands", file=sys.stderr)
print(" Type 'exit' to quit", file=sys.stderr)
print("", file=sys.stderr)
while True:
try:
command = input("dolibarr-ultra> ").strip()
if command == "exit":
break
elif command == "list":
print("Available tools:")
for i, tool in enumerate(tools, 1):
print(f" {i:2}. {tool}")
elif command == "help":
print("Commands:")
print(" list - Show all available tools")
print(" test <tool_name> - Test a specific tool")
print(" config - Show current configuration")
print(" exit - Quit the server")
print("")
print("Quick tests available:")
print(" test test_connection - Test API connection")
print(" test get_status - Get Dolibarr status")
print(" test get_users - Get first 5 users")
print(" test get_customers - Get first 5 customers")
print(" test get_products - Get first 5 products")
elif command == "config":
config = UltraSimpleConfig()
print(f"Configuration:")
print(f" URL: {config.dolibarr_url}")
print(f" API Key: {'*' * min(len(config.api_key), 10)}...")
print(f" Log Level: {config.log_level}")
elif command.startswith("test "):
tool_name = command[5:].strip()
if tool_name == "test_connection":
result = self.handle_tool_call("test_connection", {})
elif tool_name == "get_status":
result = self.handle_tool_call("get_status", {})
elif tool_name == "get_users":
result = self.handle_tool_call("get_users", {"limit": 5})
elif tool_name == "get_customers":
result = self.handle_tool_call("get_customers", {"limit": 5})
elif tool_name == "get_products":
result = self.handle_tool_call("get_products", {"limit": 5})
else:
if tool_name in tools:
print(f"Tool '{tool_name}' requires parameters. Try one of the quick tests:")
print(" test_connection, get_status, get_users, get_customers, get_products")
else:
print(f"Unknown tool: {tool_name}")
continue
print(self.format_response(result))
elif command:
print("Unknown command. Type 'help' for available commands.")
except KeyboardInterrupt:
break
except Exception as e:
print(f"Error: {e}")
print("\n👋 Goodbye!")
def main():
"""Main entry point."""
server = UltraSimpleServer("dolibarr-mcp-ultra")
server.run_interactive()
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
print("\n👋 Server stopped by user", file=sys.stderr)
sys.exit(0)
except Exception as e:
print(f"❌ Server error: {e}", file=sys.stderr)
sys.exit(1)