feat: Add comprehensive Dolibarr MCP server with full tool definitions

This commit is contained in:
latinogino
2025-07-10 12:06:07 +02:00
parent 9d54cf06b1
commit 12d3ac2107

View File

@@ -0,0 +1,678 @@
"""Professional Dolibarr MCP Server with comprehensive CRUD operations."""
import asyncio
import json
import sys
import os
# Import MCP components
from mcp.server.models import InitializationOptions
from mcp.server import NotificationOptions, Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
# Import our Dolibarr components
from .config import Config
from .dolibarr_client import DolibarrClient, DolibarrAPIError
# Create server instance
server = Server("dolibarr-mcp")
@server.list_tools()
async def handle_list_tools():
"""List all available tools."""
return [
# System & Info
Tool(
name="test_connection",
description="Test Dolibarr API connection",
inputSchema={"type": "object", "properties": {}, "additionalProperties": False}
),
Tool(
name="get_status",
description="Get Dolibarr system status and version information",
inputSchema={"type": "object", "properties": {}, "additionalProperties": False}
),
# User Management CRUD
Tool(
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
}
),
Tool(
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
}
),
Tool(
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
}
),
Tool(
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
}
),
Tool(
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
Tool(
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
}
),
Tool(
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
}
),
Tool(
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
}
),
Tool(
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
}
),
Tool(
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
Tool(
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
}
),
Tool(
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
}
),
Tool(
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
}
),
Tool(
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
}
),
Tool(
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
}
),
# Invoice Management CRUD
Tool(
name="get_invoices",
description="Get list of invoices from Dolibarr",
inputSchema={
"type": "object",
"properties": {
"limit": {"type": "integer", "description": "Maximum number of invoices to return (default: 100)", "default": 100},
"status": {"type": "string", "description": "Invoice status filter (draft, unpaid, paid, etc.)"}
},
"additionalProperties": False
}
),
Tool(
name="get_invoice_by_id",
description="Get specific invoice details by ID",
inputSchema={
"type": "object",
"properties": {
"invoice_id": {"type": "integer", "description": "Invoice ID to retrieve"}
},
"required": ["invoice_id"],
"additionalProperties": False
}
),
Tool(
name="create_invoice",
description="Create a new invoice",
inputSchema={
"type": "object",
"properties": {
"customer_id": {"type": "integer", "description": "Customer ID (socid)"},
"date": {"type": "string", "description": "Invoice date (YYYY-MM-DD)"},
"due_date": {"type": "string", "description": "Due date (YYYY-MM-DD)"},
"lines": {
"type": "array",
"description": "Invoice lines",
"items": {
"type": "object",
"properties": {
"desc": {"type": "string", "description": "Line description"},
"qty": {"type": "number", "description": "Quantity"},
"subprice": {"type": "number", "description": "Unit price"},
"total_ht": {"type": "number", "description": "Total excluding tax"},
"total_ttc": {"type": "number", "description": "Total including tax"},
"vat": {"type": "number", "description": "VAT rate"}
},
"required": ["desc", "qty", "subprice"]
}
}
},
"required": ["customer_id", "lines"],
"additionalProperties": False
}
),
Tool(
name="update_invoice",
description="Update an existing invoice",
inputSchema={
"type": "object",
"properties": {
"invoice_id": {"type": "integer", "description": "Invoice ID to update"},
"date": {"type": "string", "description": "Invoice date (YYYY-MM-DD)"},
"due_date": {"type": "string", "description": "Due date (YYYY-MM-DD)"}
},
"required": ["invoice_id"],
"additionalProperties": False
}
),
Tool(
name="delete_invoice",
description="Delete an invoice",
inputSchema={
"type": "object",
"properties": {
"invoice_id": {"type": "integer", "description": "Invoice ID to delete"}
},
"required": ["invoice_id"],
"additionalProperties": False
}
),
# Order Management CRUD
Tool(
name="get_orders",
description="Get list of orders from Dolibarr",
inputSchema={
"type": "object",
"properties": {
"limit": {"type": "integer", "description": "Maximum number of orders to return (default: 100)", "default": 100},
"status": {"type": "string", "description": "Order status filter"}
},
"additionalProperties": False
}
),
Tool(
name="get_order_by_id",
description="Get specific order details by ID",
inputSchema={
"type": "object",
"properties": {
"order_id": {"type": "integer", "description": "Order ID to retrieve"}
},
"required": ["order_id"],
"additionalProperties": False
}
),
Tool(
name="create_order",
description="Create a new order",
inputSchema={
"type": "object",
"properties": {
"customer_id": {"type": "integer", "description": "Customer ID (socid)"},
"date": {"type": "string", "description": "Order date (YYYY-MM-DD)"}
},
"required": ["customer_id"],
"additionalProperties": False
}
),
Tool(
name="update_order",
description="Update an existing order",
inputSchema={
"type": "object",
"properties": {
"order_id": {"type": "integer", "description": "Order ID to update"},
"date": {"type": "string", "description": "Order date (YYYY-MM-DD)"}
},
"required": ["order_id"],
"additionalProperties": False
}
),
Tool(
name="delete_order",
description="Delete an order",
inputSchema={
"type": "object",
"properties": {
"order_id": {"type": "integer", "description": "Order ID to delete"}
},
"required": ["order_id"],
"additionalProperties": False
}
),
# Contact Management CRUD
Tool(
name="get_contacts",
description="Get list of contacts from Dolibarr",
inputSchema={
"type": "object",
"properties": {
"limit": {"type": "integer", "description": "Maximum number of contacts to return (default: 100)", "default": 100}
},
"additionalProperties": False
}
),
Tool(
name="get_contact_by_id",
description="Get specific contact details by ID",
inputSchema={
"type": "object",
"properties": {
"contact_id": {"type": "integer", "description": "Contact ID to retrieve"}
},
"required": ["contact_id"],
"additionalProperties": False
}
),
Tool(
name="create_contact",
description="Create a new contact",
inputSchema={
"type": "object",
"properties": {
"firstname": {"type": "string", "description": "First name"},
"lastname": {"type": "string", "description": "Last name"},
"email": {"type": "string", "description": "Email address"},
"phone": {"type": "string", "description": "Phone number"},
"socid": {"type": "integer", "description": "Associated company ID"}
},
"required": ["firstname", "lastname"],
"additionalProperties": False
}
),
Tool(
name="update_contact",
description="Update an existing contact",
inputSchema={
"type": "object",
"properties": {
"contact_id": {"type": "integer", "description": "Contact ID to update"},
"firstname": {"type": "string", "description": "First name"},
"lastname": {"type": "string", "description": "Last name"},
"email": {"type": "string", "description": "Email address"},
"phone": {"type": "string", "description": "Phone number"}
},
"required": ["contact_id"],
"additionalProperties": False
}
),
Tool(
name="delete_contact",
description="Delete a contact",
inputSchema={
"type": "object",
"properties": {
"contact_id": {"type": "integer", "description": "Contact ID to delete"}
},
"required": ["contact_id"],
"additionalProperties": False
}
),
# Raw API Access
Tool(
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
}
)
]
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict):
"""Handle all 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'])
# Invoice Management
elif name == "get_invoices":
result = await client.get_invoices(
limit=arguments.get('limit', 100),
status=arguments.get('status')
)
elif name == "get_invoice_by_id":
result = await client.get_invoice_by_id(arguments['invoice_id'])
elif name == "create_invoice":
result = await client.create_invoice(**arguments)
elif name == "update_invoice":
invoice_id = arguments.pop('invoice_id')
result = await client.update_invoice(invoice_id, **arguments)
elif name == "delete_invoice":
result = await client.delete_invoice(arguments['invoice_id'])
# Order Management
elif name == "get_orders":
result = await client.get_orders(
limit=arguments.get('limit', 100),
status=arguments.get('status')
)
elif name == "get_order_by_id":
result = await client.get_order_by_id(arguments['order_id'])
elif name == "create_order":
result = await client.create_order(**arguments)
elif name == "update_order":
order_id = arguments.pop('order_id')
result = await client.update_order(order_id, **arguments)
elif name == "delete_order":
result = await client.delete_order(arguments['order_id'])
# Contact Management
elif name == "get_contacts":
result = await client.get_contacts(limit=arguments.get('limit', 100))
elif name == "get_contact_by_id":
result = await client.get_contact_by_id(arguments['contact_id'])
elif name == "create_contact":
result = await client.create_contact(**arguments)
elif name == "update_contact":
contact_id = arguments.pop('contact_id')
result = await client.update_contact(contact_id, **arguments)
elif name == "delete_contact":
result = await client.delete_contact(arguments['contact_id'])
# Raw API Access
elif name == "dolibarr_raw_api":
result = await client.dolibarr_raw_api(**arguments)
else:
result = {"error": f"Unknown tool: {name}"}
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))]
except Exception as e:
error_result = {"error": f"Tool execution failed: {str(e)}", "type": "internal_error"}
return [TextContent(type="text", text=json.dumps(error_result, indent=2))]
async def main():
"""Run the Dolibarr MCP server."""
# Quick API test using the proper client
try:
config = Config()
async with DolibarrClient(config) as client:
print("🧪 Testing Dolibarr API connection...", file=sys.stderr)
result = await client.get_status()
if 'success' in result or 'dolibarr_version' in str(result):
print("✅ Dolibarr API connection successful", file=sys.stderr)
print("🎯 Full CRUD operations available for all Dolibarr modules", file=sys.stderr)
else:
print(f"❌ API test failed: {result.get('error', 'Unknown error')}", file=sys.stderr)
return
except Exception as e:
print(f"❌ API test error: {e}", file=sys.stderr)
return
# Run server
print("🚀 Starting Professional Dolibarr MCP server...", file=sys.stderr)
print("✅ Server ready with comprehensive ERP management capabilities", file=sys.stderr)
async with stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="dolibarr-mcp",
server_version="1.0.0",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={},
),
),
)
if __name__ == "__main__":
asyncio.run(main())