dolibarr-mcp-server / app_old.py
yugapurush's picture
Day before judgement day, judas will strike in an hour
0fbe5c9
import gradio as gr
import requests
import json
import asyncio
import logging
from typing import Dict, List, Any, Optional
# Set up logging to help debug issues
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class DolibarrAPI:
base_url = "http://localhost/dolibarr/api/index.php"
def __init__(self, api_key: str):
"""
Just a constructor, sets up the api key and some other stuff
Args:
api_key (str): Dolibarr user api key
"""
self.api_key = "OKgV53jdbT1p1tKuZrB05eK9z0p9I2YX"
self.headers = {
'DOLAPIKEY': api_key,
'Content-Type': 'application/json',
'Accept': 'application/json'
}
def _request(self, method: str, endpoint: str, data: Optional[dict] = None, params: Optional[dict] = None) -> Any:
"""
basic method used for making http request to Dolibarr api
Args:
method (str): The type of HTTP - GET, POST, PUT, DELETE,etc
endpoint (str): api endpoint based on the request type - "/invoice" , "/thirdparties",etc
data (Optional[dict]) : data to be sent in request body, dictionary
params (Optional[dict]) : dictionary of params to be sent with this request (for GET)
"""
base_url = "http://localhost/dolibarr/api/index.php"
url = f"{base_url}{endpoint}"
try:
response = requests.request(method, url, headers=self.headers, json=data, params=params)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
logger.error(f"API request failed: {e}")
return {"error": f"API request failed: {str(e)}"}
except json.JSONDecodeError as e:
logger.error(f"JSON decode error: {e}")
return {"error": f"Invalid JSON response: {str(e)}"}
# --- GET REQUESTS ------
def get_req(self, endpoint: str, params: Optional[dict] = None):
return self._request('GET', endpoint, params=params)
# --- POST requests -----
def post_req(self, endpoint: str, params: dict):
return self._request("POST", endpoint, data=params)
# --- PUT requests ----
def put_req(self, endpoint: str, params: dict):
return self._request("PUT", endpoint, data=params)
# --- DELETE requests ----
def del_req(self, endpoint: str, params: Optional[dict] = None):
return self._request("DELETE", endpoint, params=params)
def dolibarr_interface(method: str, endpoint: str, api_key="OKgV53jdbT1p1tKuZrB05eK9z0p9I2YX", payload_str: str = "") -> str:
"""
To orchestrate the API call from start to finish based on simple string and dictionary inputs
Args:
method (str): http method type ( GET, POST, PUT, DELETE,etc)
endpoint (str): api endpoint, basically which api to call (/invoices, /thirdparties,/products,etc)
api_key : dolibarr api key
payload_str (str): payload as JSON string to send as request body
"""
try:
api = DolibarrAPI(api_key)
method = method.upper()
# Parse payload if provided
payload = None
if payload_str and payload_str.strip():
try:
payload = json.loads(payload_str)
except json.JSONDecodeError as e:
return json.dumps({"error": f"Invalid JSON payload: {str(e)}"}, indent=2)
if method == 'GET':
result = api.get_req(endpoint, payload)
elif method == 'POST':
if not payload:
return json.dumps({"error": "POST requests require a payload"}, indent=2)
result = api.post_req(endpoint, payload)
elif method == 'PUT':
if not payload:
return json.dumps({"error": "PUT requests require a payload"}, indent=2)
result = api.put_req(endpoint, payload)
elif method == 'DELETE':
result = api.del_req(endpoint, payload)
else:
return json.dumps({"error": f"Invalid HTTP method '{method}' selected."}, indent=2)
cleaned = clean_json_response(result)
return json.dumps(cleaned, indent=2)
except Exception as e:
logger.error(f"Unexpected error in dolibarr_interface: {e}")
return json.dumps({"error": f"Unexpected error: {str(e)}"}, indent=2)
def clean_json_response(data: Any) -> Any:
"""
Recursively clean JSON response by removing null values, empty strings, and empty collections.
Args:
data: The data to clean (can be dict, list, or primitive type)
Returns:
Cleaned data with null values and empty collections removed
"""
if isinstance(data, dict):
return {
k: clean_json_response(v)
for k, v in data.items()
if v is not None and v != "" and clean_json_response(v) is not None
}
elif isinstance(data, list):
cleaned = [clean_json_response(item) for item in data]
return [item for item in cleaned if item is not None and item != ""]
else:
return data
# Create the Gradio interface with better error handling
def create_interface():
"""Create and return the Gradio interface"""
demo = gr.Interface(
fn=dolibarr_interface,
inputs=[
gr.Dropdown(
choices=["GET", "POST", "PUT", "DELETE"],
label="HTTP Method",
value="GET"
),
gr.Dropdown(
choices=["/thirdparties", "/invoices", "/products", "/contacts", "/users"],
label="API Endpoint",
value="/thirdparties",
allow_custom_value=True
),
gr.Textbox(
label="API Key",
value="OKgV53jdbT1p1tKuZrB05eK9z0p9I2YX"
),
gr.Textbox(
label="Payload (JSON format)",
placeholder='{"ref": "PROD-001", "label": "Product Name", "price": "99.99"}'
)
],
outputs=gr.Textbox(
label="API Response",
#lines=10
),
title="Dolibarr AI Agent/Personal ERP Assistant",
description="Interact with your Dolibarr ERP system through API calls. Select method, endpoint, and provide JSON payload for POST/PUT requests.",
examples=[
["GET", "/thirdparties", "OKgV53jdbT1p1tKuZrB05eK9z0p9I2YX", ""],
["POST", "/products", "OKgV53jdbT1p1tKuZrB05eK9z0p9I2YX", '{"ref": "PROD-007", "label": "New AI-Powered Gadget", "price": "199.99", "tva_tx": "20.0"}']
]
)
return demo
# Main execution with better MCP server handling
if __name__ == '__main__':
try:
demo = create_interface()
logger.info("starting gradio app...")
demo.launch()
except Exception as e:
logger.error(f"Failed to start application: {e}")
print(f"Error starting application: {e}")
#----- Example usage ----
# Uncomment these lines to test the API functionality directly
# # Test basic API calling
# print("Testing API connection...")
# list_result = dolibarr_interface(method='GET', endpoint='/thirdparties')
# print("Thirdparties list:", list_result)
# # Test creating a product
# new_product_data = {
# "ref": "PROD-007",
# "label": "New AI-Powered Gadget",
# "price": "199.99",
# "tva_tx": "20.0"
# }
# create_result = dolibarr_interface(
# method='POST',
# endpoint='/products',
# payload_str=json.dumps(new_product_data)
# )
# print("Product creation result:", create_result)