File size: 7,943 Bytes
0fbe5c9 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 |
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) |