deel-contractor-demo/oa_client.py at main · openaccountants/deel-contractor-demo · GitHub

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
"""OpenAccountants MCP client.

JSON-RPC 2.0 to https://www.openaccountants.com/api/mcp. The MCP requires auth on
tool calls (OA_MCP_TOKEN) for live mode; without one it returns bundled sample
responses that mirror the live shape so the demo runs end to end.

  - start(intent, jurisdiction)  -> which verified skill governs the question
  - get_skill(slug)              -> the contractor-payment rules + tier + named verifier
"""

from __future__ import annotations

import json
import os
import urllib.request
import urllib.error

MCP_URL = os.environ.get("OA_MCP_URL", "https://www.openaccountants.com/api/mcp")
MCP_TOKEN = os.environ.get("OA_MCP_TOKEN")


class OAClient:
    def __init__(self, token: str | None = MCP_TOKEN, url: str = MCP_URL):
        self.token, self.url, self._id = token, url, 0

    @property
    def live(self) -> bool:
        return bool(self.token)

    def start(self, intent: str, jurisdiction: str) -> dict:
        if not self.live:
            return _MOCK_START.get(jurisdiction.upper(), _MOCK_START["_DEFAULT"])
        return self._call("start", {"intent": intent, "jurisdiction": jurisdiction})

    def get_skill(self, slug: str) -> dict:
        if not self.live:
            return _MOCK_SKILL.get(slug, _MOCK_SKILL["_DEFAULT"])
        return self._call("get_skill", {"slug": slug})

    def _call(self, tool: str, arguments: dict) -> dict:
        self._id += 1
        body = json.dumps({"jsonrpc": "2.0", "id": self._id, "method": "tools/call",
                           "params": {"name": tool, "arguments": arguments}}).encode()
        headers = {"Content-Type": "application/json", "Accept": "application/json"}
        if self.token:
            headers["Authorization"] = f"Bearer {self.token}"  # scheme TBD; swap if OA uses apikey
        req = urllib.request.Request(self.url, data=body, headers=headers)
        try:
            with urllib.request.urlopen(req, timeout=30) as resp:
                payload = json.load(resp)
        except urllib.error.HTTPError as e:
            raise RuntimeError(f"OA MCP HTTP {e.code}: {e.read().decode()[:300]}") from e
        if "error" in payload and payload["error"]:
            raise RuntimeError(f"OA MCP error: {payload['error']}")
        result = payload.get("result", {})
        if isinstance(result.get("structuredContent"), (dict, list)):
            return result["structuredContent"]
        return json.loads((result.get("content") or [{}])[0].get("text", "{}"))


# --- Bundled sample responses (mirror live MCP shape) ----------------------
# Verifier is the real OpenAccountants US lead. Live, every value comes from get_skill.

_MOCK_START = {
    "US": {"jurisdiction": "US", "skills_to_load": ["us-contractor-payments"],
           "next_action": "Load the skill, then check each contractor's tax obligations."},
    "_DEFAULT": {"jurisdiction": "US", "skills_to_load": ["us-contractor-payments"],
                 "next_action": "Load the skill, then check each contractor's tax obligations."},
}

_MOCK_SKILL = {
    "us-contractor-payments": {
        "slug": "us-contractor-payments",
        "name": "US contractor payments (1099 / W-8BEN / classification)",
        "jurisdiction": "US", "tier": 1, "verifier": "Amir Pelinkovic (US lead)",
        "rules": {
            # Illustrative — production reads these from the skill.
            "form_1099_threshold": 600.0,    # 1099-NEC required for US persons paid > $600
            "us_person_form": "W-9",
            "foreign_person_form": "W-8BEN",
            "us_source_withholding": 0.30,   # default FDAP/US-source withholding absent a treaty rate
        },
        "source": "https://www.openaccountants.com/skills/us-contractor-payments",
    },
    "_DEFAULT": {"slug": "us-contractor-payments", "name": "Contractor payments",
                 "jurisdiction": "US", "tier": 2, "verifier": None, "rules": {},
                 "source": "https://www.openaccountants.com/skills"},
}