Add ProxyAwareTransport to fix xmlrpc proxy bypass in Cowork sandbox

Routes all XML-RPC calls through the system HTTPS proxy (HTTPS_PROXY env var)
using urllib.request.build_opener(ProxyHandler()) instead of raw socket
connections that bypass the proxy. Also adds urllib.request and urllib.error
imports.
This commit is contained in:
2026-03-31 08:53:15 -05:00
parent be6266b2c8
commit 19ed0844f9
+27 -2
View File
@@ -9,6 +9,8 @@ Purchase, Inventory, Employees, and Knowledge Templates.
import os
import re
import xmlrpc.client
import urllib.request
import urllib.error
from typing import Optional
from mcp.server.fastmcp import FastMCP
@@ -18,6 +20,28 @@ ODOO_DB = os.environ.get("ODOO_DB", "mpmedia-odoo-sh-main-13285275")
ODOO_USERNAME = os.environ.get("ODOO_USERNAME", "bgilliom@mpmedia.tv")
ODOO_API_KEY = os.environ.get("ODOO_API_KEY", "")
# ── Proxy-aware XML-RPC transport ─────────────────────────────────────────────
class ProxyAwareTransport(xmlrpc.client.SafeTransport):
"""Routes xmlrpc through the system HTTPS proxy (respects HTTPS_PROXY env var)."""
def request(self, host, handler, request_body, verbose=False):
url = f"https://{host}{handler}"
headers = {
"Content-Type": "text/xml",
"Accept-Encoding": "identity",
"User-Agent": "xmlrpc-odoo-mpm/1.0",
}
req = urllib.request.Request(url, request_body, headers)
opener = urllib.request.build_opener(urllib.request.ProxyHandler())
try:
with opener.open(req, timeout=30) as resp:
return self.parse_response(resp)
except urllib.error.HTTPError as e:
raise xmlrpc.client.ProtocolError(url, e.code, e.msg, dict(e.headers))
except urllib.error.URLError as e:
raise xmlrpc.client.ProtocolError(url, 0, str(e.reason), {})
_proxy_transport = ProxyAwareTransport()
# ── Odoo client ───────────────────────────────────────────────────────────────
_uid: Optional[int] = None
_models = None
@@ -26,11 +50,11 @@ def _connect():
global _uid, _models
if _uid is not None:
return
common = xmlrpc.client.ServerProxy(f"{ODOO_URL}/xmlrpc/2/common")
common = xmlrpc.client.ServerProxy(f"{ODOO_URL}/xmlrpc/2/common", transport=_proxy_transport)
_uid = common.authenticate(ODOO_DB, ODOO_USERNAME, ODOO_API_KEY, {})
if not _uid:
raise RuntimeError("Odoo authentication failed. Check ODOO_USERNAME and ODOO_API_KEY.")
_models = xmlrpc.client.ServerProxy(f"{ODOO_URL}/xmlrpc/2/object")
_models = xmlrpc.client.ServerProxy(f"{ODOO_URL}/xmlrpc/2/object", transport=_proxy_transport)
def _call(model: str, method: str, args=None, kwargs=None):
_connect()
@@ -98,6 +122,7 @@ def get_product(product_id: int) -> dict:
def get_product_stock(product_id: int) -> list:
"""Get current stock quantities for a product (by product.template ID)
across all internal locations."""
# Get all product.product IDs under this template
variants = _search_read("product.product",
[["product_tmpl_id", "=", product_id]],
["id", "display_name"])