mirror of
https://github.com/latinogino/dolibarr-mcp.git
synced 2026-05-01 13:55:35 +02:00
Fix URL construction in test script - remove duplicate path segments
This commit is contained in:
@@ -5,6 +5,7 @@ import os
|
|||||||
import sys
|
import sys
|
||||||
import asyncio
|
import asyncio
|
||||||
import aiohttp
|
import aiohttp
|
||||||
|
import json
|
||||||
from dotenv import load_dotenv
|
from dotenv import load_dotenv
|
||||||
|
|
||||||
# Load environment variables
|
# Load environment variables
|
||||||
@@ -28,20 +29,18 @@ async def test_dolibarr_connection():
|
|||||||
print(" Please set DOLIBARR_URL and DOLIBARR_API_KEY in .env file")
|
print(" Please set DOLIBARR_URL and DOLIBARR_API_KEY in .env file")
|
||||||
return False
|
return False
|
||||||
|
|
||||||
# Ensure URL format is correct
|
# Clean base URL - remove trailing slash if present
|
||||||
if not base_url.endswith('/api/index.php'):
|
base_url = base_url.rstrip('/')
|
||||||
if base_url.endswith('/'):
|
|
||||||
base_url = base_url + 'api/index.php'
|
|
||||||
else:
|
|
||||||
base_url = base_url + '/api/index.php'
|
|
||||||
|
|
||||||
# Test different endpoints
|
# Test different endpoints
|
||||||
endpoints_to_test = [
|
endpoints_to_test = [
|
||||||
"/status",
|
"status", # API status
|
||||||
"/explorer/swagger", # Swagger documentation
|
"users", # Users list
|
||||||
"/setup/modules", # Module list
|
"thirdparties", # Customers/Suppliers
|
||||||
"/users", # Users list
|
"products", # Products
|
||||||
"/thirdparties", # Customers/Suppliers
|
"invoices", # Invoices
|
||||||
|
"orders", # Orders
|
||||||
|
"contacts", # Contacts
|
||||||
]
|
]
|
||||||
|
|
||||||
# Headers for Dolibarr API
|
# Headers for Dolibarr API
|
||||||
@@ -54,112 +53,173 @@ async def test_dolibarr_connection():
|
|||||||
print("🧪 Testing Dolibarr API endpoints:")
|
print("🧪 Testing Dolibarr API endpoints:")
|
||||||
print("=" * 50)
|
print("=" * 50)
|
||||||
|
|
||||||
|
working_endpoints = []
|
||||||
|
|
||||||
async with aiohttp.ClientSession() as session:
|
async with aiohttp.ClientSession() as session:
|
||||||
for endpoint in endpoints_to_test:
|
for endpoint in endpoints_to_test:
|
||||||
# Clean endpoint
|
url = f"{base_url}/{endpoint}"
|
||||||
endpoint = endpoint.lstrip('/')
|
|
||||||
|
|
||||||
# Build full URL
|
try:
|
||||||
if endpoint in ["status", "explorer/swagger"]:
|
print(f"\n📍 Testing: {endpoint}")
|
||||||
# These endpoints might be at root level
|
print(f" URL: {url}")
|
||||||
test_urls = [
|
|
||||||
f"{base_url}/{endpoint}",
|
async with session.get(url, headers=headers, timeout=10, ssl=False) as response:
|
||||||
f"{base_url.replace('/api/index.php', '')}/{endpoint}",
|
status = response.status
|
||||||
f"{base_url.replace('/api/index.php', '/api')}/{endpoint}"
|
text = await response.text()
|
||||||
]
|
|
||||||
else:
|
|
||||||
test_urls = [f"{base_url}/{endpoint}"]
|
|
||||||
|
|
||||||
success = False
|
|
||||||
for url in test_urls:
|
|
||||||
try:
|
|
||||||
print(f"\n📍 Testing: {url}")
|
|
||||||
|
|
||||||
async with session.get(url, headers=headers, timeout=10) as response:
|
print(f" Status: {status}")
|
||||||
status = response.status
|
|
||||||
text = await response.text()
|
if status == 200:
|
||||||
|
print(f" ✅ Success!")
|
||||||
print(f" Status: {status}")
|
working_endpoints.append(endpoint)
|
||||||
|
try:
|
||||||
if status == 200:
|
data = json.loads(text)
|
||||||
print(f" ✅ Success!")
|
if isinstance(data, dict):
|
||||||
if text:
|
print(f" Response keys: {list(data.keys())[:5]}")
|
||||||
# Try to parse response
|
elif isinstance(data, list):
|
||||||
try:
|
print(f" Response: List with {len(data)} items")
|
||||||
import json
|
if len(data) > 0 and isinstance(data[0], dict):
|
||||||
data = json.loads(text)
|
print(f" First item keys: {list(data[0].keys())[:5]}")
|
||||||
if isinstance(data, dict):
|
except:
|
||||||
print(f" Response keys: {list(data.keys())[:5]}")
|
print(f" Response preview: {text[:100]}...")
|
||||||
elif isinstance(data, list):
|
elif status == 401:
|
||||||
print(f" Response: List with {len(data)} items")
|
print(f" ❌ Authentication failed - check API key")
|
||||||
except:
|
elif status == 403:
|
||||||
print(f" Response preview: {text[:100]}...")
|
print(f" ❌ Access forbidden - check permissions")
|
||||||
success = True
|
elif status == 404:
|
||||||
break
|
print(f" ❌ Endpoint not found")
|
||||||
elif status == 401:
|
elif status == 501:
|
||||||
print(f" ❌ Authentication failed - check API key")
|
print(f" ⚠️ API module not found - endpoint may not be available")
|
||||||
elif status == 403:
|
if text:
|
||||||
print(f" ❌ Access forbidden - check permissions")
|
|
||||||
elif status == 404:
|
|
||||||
print(f" ❌ Endpoint not found")
|
|
||||||
else:
|
|
||||||
print(f" ⚠️ Unexpected status: {status}")
|
|
||||||
print(f" Response: {text[:200]}...")
|
print(f" Response: {text[:200]}...")
|
||||||
|
else:
|
||||||
except aiohttp.ClientError as e:
|
print(f" ⚠️ Unexpected status: {status}")
|
||||||
print(f" ❌ Connection error: {e}")
|
if text:
|
||||||
except Exception as e:
|
print(f" Response: {text[:200]}...")
|
||||||
print(f" ❌ Unexpected error: {e}")
|
|
||||||
|
except aiohttp.ClientError as e:
|
||||||
if success:
|
print(f" ❌ Connection error: {type(e).__name__}: {e}")
|
||||||
print(f" ✨ Endpoint {endpoint} is working!")
|
except Exception as e:
|
||||||
else:
|
print(f" ❌ Unexpected error: {type(e).__name__}: {e}")
|
||||||
print(f" ⚠️ Endpoint {endpoint} not accessible")
|
|
||||||
|
|
||||||
print("\n" + "=" * 50)
|
print("\n" + "=" * 50)
|
||||||
print("\n📝 Recommendations:")
|
|
||||||
print("1. Check if Dolibarr Web Services API REST module is enabled")
|
|
||||||
print("2. Verify API key is correct and has proper permissions")
|
|
||||||
print("3. Ensure URL format: https://domain.com/api/index.php")
|
|
||||||
|
|
||||||
return True
|
if working_endpoints:
|
||||||
|
print("\n✨ Working endpoints:")
|
||||||
|
for endpoint in working_endpoints:
|
||||||
|
print(f" - {endpoint}")
|
||||||
|
else:
|
||||||
|
print("\n⚠️ No endpoints are working!")
|
||||||
|
|
||||||
|
return len(working_endpoints) > 0
|
||||||
|
|
||||||
|
|
||||||
async def test_swagger_discovery():
|
async def test_swagger_endpoint():
|
||||||
"""Try to discover API endpoints via Swagger/OpenAPI."""
|
"""Test Swagger/Explorer endpoint specifically."""
|
||||||
|
|
||||||
base_url = os.getenv("DOLIBARR_URL", "")
|
base_url = os.getenv("DOLIBARR_URL", "").rstrip('/')
|
||||||
api_key = os.getenv("DOLIBARR_API_KEY", "")
|
api_key = os.getenv("DOLIBARR_API_KEY", "")
|
||||||
|
|
||||||
if not base_url or not api_key:
|
if not base_url or not api_key:
|
||||||
return
|
return
|
||||||
|
|
||||||
print("\n🔍 Attempting Swagger/OpenAPI discovery:")
|
print("\n🔍 Testing Swagger/Explorer endpoints:")
|
||||||
print("=" * 50)
|
print("=" * 50)
|
||||||
|
|
||||||
|
# Swagger endpoints to test
|
||||||
|
swagger_endpoints = [
|
||||||
|
"explorer",
|
||||||
|
"explorer/index.html",
|
||||||
|
f"explorer/swagger.json?DOLAPIKEY={api_key}",
|
||||||
|
]
|
||||||
|
|
||||||
headers = {
|
headers = {
|
||||||
"DOLAPIKEY": api_key,
|
"DOLAPIKEY": api_key,
|
||||||
|
"Accept": "application/json, text/html, */*"
|
||||||
|
}
|
||||||
|
|
||||||
|
async with aiohttp.ClientSession() as session:
|
||||||
|
for endpoint in swagger_endpoints:
|
||||||
|
url = f"{base_url}/{endpoint}"
|
||||||
|
|
||||||
|
try:
|
||||||
|
print(f"\nTesting: {url}")
|
||||||
|
|
||||||
|
async with session.get(url, headers=headers, timeout=10, ssl=False) as response:
|
||||||
|
status = response.status
|
||||||
|
content_type = response.headers.get('Content-Type', '')
|
||||||
|
|
||||||
|
print(f" Status: {status}")
|
||||||
|
print(f" Content-Type: {content_type}")
|
||||||
|
|
||||||
|
if status == 200:
|
||||||
|
print(f" ✅ Found!")
|
||||||
|
|
||||||
|
# If it's the swagger.json, try to parse it
|
||||||
|
if 'swagger.json' in endpoint:
|
||||||
|
text = await response.text()
|
||||||
|
try:
|
||||||
|
swagger_data = json.loads(text)
|
||||||
|
if 'paths' in swagger_data:
|
||||||
|
print(f" Available API endpoints:")
|
||||||
|
for path in list(swagger_data['paths'].keys())[:10]:
|
||||||
|
print(f" - {path}")
|
||||||
|
if len(swagger_data['paths']) > 10:
|
||||||
|
print(f" ... and {len(swagger_data['paths']) - 10} more")
|
||||||
|
except:
|
||||||
|
print(f" Could not parse Swagger JSON")
|
||||||
|
else:
|
||||||
|
text = await response.text()
|
||||||
|
print(f" Response preview: {text[:100]}...")
|
||||||
|
|
||||||
|
except Exception as e:
|
||||||
|
print(f" Error: {type(e).__name__}: {e}")
|
||||||
|
|
||||||
|
|
||||||
|
async def test_login_endpoint():
|
||||||
|
"""Test the login endpoint to get a session token."""
|
||||||
|
|
||||||
|
base_url = os.getenv("DOLIBARR_URL", "").rstrip('/')
|
||||||
|
api_key = os.getenv("DOLIBARR_API_KEY", "")
|
||||||
|
|
||||||
|
if not base_url or not api_key:
|
||||||
|
return
|
||||||
|
|
||||||
|
print("\n🔐 Testing Login endpoint:")
|
||||||
|
print("=" * 50)
|
||||||
|
|
||||||
|
# Test with API key in header (standard method)
|
||||||
|
url = f"{base_url}/login"
|
||||||
|
|
||||||
|
headers = {
|
||||||
|
"DOLAPIKEY": api_key,
|
||||||
|
"Content-Type": "application/json",
|
||||||
"Accept": "application/json"
|
"Accept": "application/json"
|
||||||
}
|
}
|
||||||
|
|
||||||
# Possible Swagger/OpenAPI URLs
|
|
||||||
swagger_urls = [
|
|
||||||
base_url.replace('/api/index.php', '/api/explorer'),
|
|
||||||
base_url.replace('/api/index.php', '/api/swagger.json'),
|
|
||||||
base_url.replace('/api/index.php', '/api/openapi.json'),
|
|
||||||
base_url.replace('/api/index.php', '/api/v2/swagger.json'),
|
|
||||||
]
|
|
||||||
|
|
||||||
async with aiohttp.ClientSession() as session:
|
async with aiohttp.ClientSession() as session:
|
||||||
for url in swagger_urls:
|
try:
|
||||||
try:
|
print(f"Testing: {url}")
|
||||||
print(f"Testing: {url}")
|
print(f"Method: GET with DOLAPIKEY header")
|
||||||
async with session.get(url, headers=headers, timeout=5) as response:
|
|
||||||
if response.status == 200:
|
async with session.get(url, headers=headers, timeout=10, ssl=False) as response:
|
||||||
print(f"✅ Found API documentation at: {url}")
|
status = response.status
|
||||||
break
|
text = await response.text()
|
||||||
except:
|
|
||||||
pass
|
print(f" Status: {status}")
|
||||||
|
|
||||||
|
if status == 200:
|
||||||
|
print(f" ✅ Authentication successful!")
|
||||||
|
try:
|
||||||
|
data = json.loads(text)
|
||||||
|
print(f" Response: {json.dumps(data, indent=2)}")
|
||||||
|
except:
|
||||||
|
print(f" Response: {text}")
|
||||||
|
else:
|
||||||
|
print(f" Response: {text[:200]}...")
|
||||||
|
|
||||||
|
except Exception as e:
|
||||||
|
print(f" Error: {type(e).__name__}: {e}")
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
@@ -167,8 +227,23 @@ if __name__ == "__main__":
|
|||||||
print("================================\n")
|
print("================================\n")
|
||||||
|
|
||||||
try:
|
try:
|
||||||
asyncio.run(test_dolibarr_connection())
|
# Run tests
|
||||||
asyncio.run(test_swagger_discovery())
|
success = asyncio.run(test_dolibarr_connection())
|
||||||
|
asyncio.run(test_swagger_endpoint())
|
||||||
|
asyncio.run(test_login_endpoint())
|
||||||
|
|
||||||
|
print("\n" + "=" * 50)
|
||||||
|
print("\n📝 Summary:")
|
||||||
|
if success:
|
||||||
|
print(" ✅ API connection is working!")
|
||||||
|
print(" You can proceed with MCP server implementation.")
|
||||||
|
else:
|
||||||
|
print(" ⚠️ API connection issues detected.")
|
||||||
|
print(" Please check:")
|
||||||
|
print(" 1. Dolibarr Web Services API REST module is enabled")
|
||||||
|
print(" 2. API key is correct and has proper permissions")
|
||||||
|
print(" 3. URL format is: https://domain.com/api/index.php/")
|
||||||
|
|
||||||
except KeyboardInterrupt:
|
except KeyboardInterrupt:
|
||||||
print("\n\n👋 Test cancelled by user")
|
print("\n\n👋 Test cancelled by user")
|
||||||
sys.exit(0)
|
sys.exit(0)
|
||||||
|
|||||||
Reference in New Issue
Block a user