Release v0.2.37

feat: optional tool-registry proxy connector

docs/registry_proxy_integration.md

@@ -0,0 +1,115 @@
+# Tool-Registry Proxy Connector
+
+An **optional, bring-your-own-account** connector that lets any PraisonAI agent
+discover and call a large catalogue of third-party API endpoints (SEO,
+enrichment, social data, scraping, ads) through a single proxy token — with
+**credentials injected server-side by the registry, never held by the agent**.
+
+PraisonAI ships **no registry code, no bundled binary and no default endpoint**.
+You supply your own account token or a self-hosted base URL. The connector is
+disabled unless `TOOL_PROXY_URL` is configured.
+
+## Overview
+
+The connector mirrors PraisonAI's deferred tool-search bridge
+(`tool_search` / `tool_describe` / `tool_call`) with three functions:
+
+| Function | Cost | Purpose |
+| --- | --- | --- |
+| `registry_search(query)` | free | capability search over the catalogue |
+| `registry_describe(tool_id)` | free | params, price-per-call, example response |
+| `registry_call(tool_id, params)` | paid | invoke via proxy; credential injected server-side |
+
+## Installation
+
+```bash
+pip install "praisonai-tools[registry-proxy]"
+```
+
+## Setup (3 lines)
+
+```bash
+export TOOL_PROXY_URL="https://your-registry.example.com"   # or a self-hosted instance
+export TOOL_PROXY_TOKEN="your-proxy-token"
+```
+
+```python
+from praisonaiagents import Agent
+from praisonai_tools.registry_proxy import (
+    registry_search, registry_describe, registry_call,
+)
+
+agent = Agent(
+    instructions="SEO analyst",
+    tools=[registry_search, registry_describe, registry_call],
+)
+agent.start("Find the top backlink sources for example.com and summarise")
+# → registry_search("backlinks for a domain") → describe → call
+# one token, no vendor keys in the agent
+```
+
+## Spend safety
+
+`registry_call` accepts optional budget guards. When a configured budget would
+be exceeded, the call is **denied and reported** as a tool-result error (never
+raised as an exception):
+
+```python
+registry_call(
+    "seo.backlinks",
+    {"domain": "example.com"},
+    max_cost_per_call=0.10,     # deny if price-per-call exceeds this
+    max_session_spend=5.00,     # deny if cumulative spend would exceed this
+)
+```
+
+Per-call price is read from `registry_describe`, so guards work without any
+vendor-specific configuration. Guards **fail closed**: if a budget is set but
+the price cannot be validated, the call is denied. Cumulative `max_session_spend`
+persists across separate `registry_call` invocations within the same process,
+scoped per `(proxy_url, token)` so distinct accounts never share a budget.
+
+## Credential safety
+
+The proxy token is only sent to the endpoint it was configured for. If a caller
+(or a prompt-injected agent) supplies an explicit `proxy_url` without a matching
+`token`, the ambient `TOOL_PROXY_TOKEN` is **not** attached — preventing the
+token from being exfiltrated to an untrusted endpoint. For normal use, leave
+`proxy_url`/`token` unset and configure `TOOL_PROXY_URL`/`TOOL_PROXY_TOKEN` via
+the environment.
+
+## Error taxonomy
+
+All failures are returned as `{"error": ...}` tool results:
+
+- Authentication failed (HTTP 401/403)
+- Insufficient balance (HTTP 402)
+- Upstream 4xx/5xx passed through
+- Request timeout
+- `httpx` not installed → install hint
+
+## Tier 0 — Skills directory compatibility (zero code)
+
+Registries that expose `skill install <name>` install skills into
+`./.claude/skills/`. PraisonAI **already scans** this directory, so
+registry-installed skills are discoverable by PraisonAI agents today with no
+additional code:
+
+- `praisonaiagents/skills/discovery.py` — `.claude/skills` compatibility scan
+  plus an ancestor walk for monorepos.
+
+## Tier 2 — Environment / CLI passthrough (config-only)
+
+A registry's `shell` / `run <vendor> -- args` inject vendor-CLI credentials at
+execution time. Using a PraisonAI environment definition, install the registry
+CLI and log in via a token env var in `setup:`; the agent's `execute_command`
+then has authenticated vendor CLIs (e.g. `run gh -- pr list`) without any key in
+the container. Sandbox / board workers inherit the same via the environment
+definition.
+
+## Licensing note
+
+Some registries are Apache-2.0 **with additional terms** prohibiting embedding
+in commercially distributed products without permission. This connector is
+therefore **optional and bring-your-own-account only**: it ships no registry
+code and no default endpoint — the same posture as other BYO-account connectors.

praisonai_tools/registry_proxy/__init__.py

@@ -0,0 +1,43 @@
+"""Tool-registry proxy connector for PraisonAI Agents.
+
+Optional bring-your-own-account connector that lets an agent discover and call
+a large catalogue of third-party API endpoints through a single proxy token,
+with credentials injected server-side by the registry (never held by the agent).
+
+Mirrors PraisonAI's deferred tool-search bridge:
+    registry_search(query)      -> capability search over the catalogue
+    registry_describe(tool_id)  -> params, price-per-call, example response
+    registry_call(tool_id, ...) -> invoke via proxy (credential injected server-side)
+
+The connector is disabled unless ``TOOL_PROXY_URL`` is configured. PraisonAI
+ships no registry code, no bundled binary and no default endpoint - the user
+supplies their own account token or self-hosted base URL.
+
+Usage:
+    from praisonai_tools.registry_proxy import (
+        registry_search, registry_describe, registry_call,
+    )
+
+    agent = Agent(
+        instructions="SEO analyst",
+        tools=[registry_search, registry_describe, registry_call],
+    )
+
+Environment Variables:
+    TOOL_PROXY_URL:   base URL of the tool registry (unset -> connector disabled)
+    TOOL_PROXY_TOKEN: proxy token sent as the auth header
+"""
+
+from .registry_proxy import (
+    RegistryProxyTool,
+    registry_search,
+    registry_describe,
+    registry_call,
+)
+
+__all__ = [
+    "RegistryProxyTool",
+    "registry_search",
+    "registry_describe",
+    "registry_call",
+]

praisonai_tools/registry_proxy/registry_proxy.py

@@ -0,0 +1,345 @@
+"""Tool-registry proxy connector implementation.
+
+Three agent-callable functions over a registry's plain-HTTP surface:
+
+- ``registry_search(query)``     free capability search over the catalogue
+- ``registry_describe(tool_id)`` free read: params, price-per-call, example
+- ``registry_call(tool_id, ...)``paid invoke through the proxy; the registry
+  injects the upstream vendor credential server-side.
+
+Design notes:
+- ``httpx`` is lazy-imported so the connector adds zero import-time overhead and
+  no hard dependency when unconfigured.
+- The connector is disabled unless ``TOOL_PROXY_URL`` is set.
+- Errors (auth / insufficient balance / upstream 4xx / timeout) are returned as
+  ``{"error": ...}`` tool results, never raised as exceptions.
+- ``registry_call`` supports optional per-call and per-session spend guards; the
+  price is read from ``registry_describe`` and the call is denied + reported when
+  a configured budget would be exceeded.
+"""
+
+import os
+import logging
+from typing import Any, Dict, Optional
+
+from praisonai_tools.tools.base import BaseTool
+from praisonai_tools.tools.decorator import tool
+
+logger = logging.getLogger(__name__)
+
+_INSTALL_HINT = "httpx not installed. Install with: pip install 'praisonai-tools[registry-proxy]'"
+
+# Cumulative spend is tracked per (proxy_url, token) so that ``max_session_spend``
+# persists across the short-lived connector instances created by the decorated
+# functions, without mixing budgets across distinct accounts/endpoints.
+_SESSION_SPEND: Dict[tuple, float] = {}
+
+
+class RegistryProxyTool(BaseTool):
+    """Connector for a self-hostable tool registry exposing a plain-HTTP proxy."""
+
+    name = "registry_proxy"
+    description = (
+        "Discover and call third-party API endpoints through a tool registry "
+        "proxy; credentials are injected server-side, never held by the agent."
+    )
+
+    def __init__(
+        self,
+        proxy_url: Optional[str] = None,
+        token: Optional[str] = None,
+        timeout: float = 60.0,
+        max_cost_per_call: Optional[float] = None,
+        max_session_spend: Optional[float] = None,
+    ):
+        url_from_env = proxy_url is None
+        self.proxy_url = (proxy_url or os.getenv("TOOL_PROXY_URL") or "").rstrip("/")
+        # Security: only fall back to the environment token when the URL is also
+        # env-derived (trusted config). An agent-supplied ``proxy_url`` must never
+        # receive the ambient ``TOOL_PROXY_TOKEN`` - otherwise a prompt-injected
+        # agent could exfiltrate the token to an attacker-controlled endpoint.
+        if token is not None:
+            self.token = token
+        elif url_from_env:
+            self.token = os.getenv("TOOL_PROXY_TOKEN")
+        else:
+            self.token = None
+        self.timeout = timeout
+        self.max_cost_per_call = max_cost_per_call
+        self.max_session_spend = max_session_spend
+        super().__init__()
+
+    @property
+    def _spend_key(self) -> tuple:
+        return (self.proxy_url, self.token)
+
+    @property
+    def _session_spend(self) -> float:
+        return _SESSION_SPEND.get(self._spend_key, 0.0)
+
+    @_session_spend.setter
+    def _session_spend(self, value: float) -> None:
+        _SESSION_SPEND[self._spend_key] = value
+
+    def run(self, action: str, **kwargs: Any) -> Dict[str, Any]:
+        """Dispatch to search/describe/call.
+
+        Args:
+            action: One of ``"search"``, ``"describe"`` or ``"call"``.
+            **kwargs: Arguments forwarded to the selected operation.
+        """
+        if action == "search":
+            return self.search(kwargs.get("query", ""))
+        if action == "describe":
+            return self.describe(kwargs.get("tool_id", ""))
+        if action == "call":
+            return self.call(kwargs.get("tool_id", ""), kwargs.get("params"))
+        return {"error": f"Unknown action '{action}'. Use search, describe or call."}
+
+    def _headers(self) -> Dict[str, str]:
+        headers = {"Content-Type": "application/json"}
+        if self.token:
+            headers["Authorization"] = f"Bearer {self.token}"
+        return headers
+
+    def _require_config(self) -> Optional[Dict[str, Any]]:
+        if not self.proxy_url:
+            return {
+                "error": (
+                    "Tool registry proxy is not configured. Set TOOL_PROXY_URL "
+                    "(and optionally TOOL_PROXY_TOKEN) to enable the connector."
+                )
+            }
+        return None
+
+    @staticmethod
+    def _http_error(exc: Any) -> Dict[str, Any]:
+        response = getattr(exc, "response", None)
+        status = getattr(response, "status_code", None)
+        text = getattr(response, "text", "") if response is not None else str(exc)
+        if status == 401 or status == 403:
+            return {"error": f"Authentication failed (HTTP {status}): {text}"}
+        if status == 402:
+            return {"error": f"Insufficient balance (HTTP {status}): {text}"}
+        return {"error": f"HTTP {status}: {text}"}
+
+    def search(self, query: str) -> Dict[str, Any]:
+        """Search the registry catalogue by capability."""
+        missing = self._require_config()
+        if missing:
+            return missing
+        if not query:
+            return {"error": "query is required"}
+        try:
+            import httpx
+        except ImportError:
+            return {"error": _INSTALL_HINT}
+        try:
+            with httpx.Client(timeout=self.timeout) as client:
+                response = client.get(
+                    f"{self.proxy_url}/catalog/search",
+                    params={"q": query},
+                    headers=self._headers(),
+                )
+                response.raise_for_status()
+                return response.json()
+        except httpx.TimeoutException:
+            return {"error": f"Registry search timed out after {self.timeout} seconds"}
+        except httpx.HTTPStatusError as exc:
+            return self._http_error(exc)
+        except Exception as exc:  # noqa: BLE001 - surface as tool-result error
+            logger.error("registry search error: %s", exc)
+            return {"error": str(exc)}
+
+    def describe(self, tool_id: str) -> Dict[str, Any]:
+        """Describe one catalogue entry: params, price-per-call, example."""
+        missing = self._require_config()
+        if missing:
+            return missing
+        if not tool_id:
+            return {"error": "tool_id is required"}
+        try:
+            import httpx
+        except ImportError:
+            return {"error": _INSTALL_HINT}
+        try:
+            with httpx.Client(timeout=self.timeout) as client:
+                response = client.get(
+                    f"{self.proxy_url}/catalog/{tool_id}",
+                    headers=self._headers(),
+                )
+                response.raise_for_status()
+                return response.json()
+        except httpx.TimeoutException:
+            return {"error": f"Registry describe timed out after {self.timeout} seconds"}
+        except httpx.HTTPStatusError as exc:
+            return self._http_error(exc)
+        except Exception as exc:  # noqa: BLE001 - surface as tool-result error
+            logger.error("registry describe error: %s", exc)
+            return {"error": str(exc)}
+
+    def _check_budget(self, tool_id: str):
+        """Deny + report if a configured spend guard would be exceeded.
+
+        Returns ``(denial, price)``: ``denial`` is a ``{"error": ...}`` dict when
+        the call must be rejected (else ``None``); ``price`` is the validated
+        per-call price when a guard applies (else ``None``).
+        """
+        if self.max_cost_per_call is None and self.max_session_spend is None:
+            return None, None
+        described = self.describe(tool_id)
+        if isinstance(described, dict) and "error" in described:
+            return described, None
+        price = described.get("price_per_call") if isinstance(described, dict) else None
+        try:
+            price = float(price)
+        except (TypeError, ValueError):
+            # Fail closed: a spend guard is configured but the price cannot be
+            # validated, so we cannot guarantee the budget - deny the call.
+            return {
+                "error": (
+                    "Denied: price_per_call is missing or invalid, cannot enforce "
+                    "the configured spend guard."
+                )
+            }, None
+        if self.max_cost_per_call is not None and price > self.max_cost_per_call:
+            return {
+                "error": (
+                    f"Denied: price per call {price} exceeds max_cost_per_call "
+                    f"{self.max_cost_per_call}."
+                )
+            }, price
+        if (
+            self.max_session_spend is not None
+            and self._session_spend + price > self.max_session_spend
+        ):
+            return {
+                "error": (
+                    f"Denied: session spend would reach "
+                    f"{self._session_spend + price}, exceeding max_session_spend "
+                    f"{self.max_session_spend}."
+                )
+            }, price
+        return None, price
+
+    def call(self, tool_id: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+        """Invoke an endpoint through the proxy (credential injected server-side)."""
+        missing = self._require_config()
+        if missing:
+            return missing
+        if not tool_id:
+            return {"error": "tool_id is required"}
+        denied, quoted_price = self._check_budget(tool_id)
+        if denied:
+            return denied
+        try:
+            import httpx
+        except ImportError:
+            return {"error": _INSTALL_HINT}
+        try:
+            with httpx.Client(timeout=self.timeout) as client:
+                response = client.post(
+                    f"{self.proxy_url}/call/{tool_id}",
+                    json=params or {},
+                    headers=self._headers(),
+                )
+                response.raise_for_status()
+                result = response.json()
+        except httpx.TimeoutException:
+            return {"error": f"Registry call timed out after {self.timeout} seconds"}
+        except httpx.HTTPStatusError as exc:
+            return self._http_error(exc)
+        except Exception as exc:  # noqa: BLE001 - surface as tool-result error
+            logger.error("registry call error: %s", exc)
+            return {"error": str(exc)}
+
+        # Record cumulative spend. Prefer the charge reported by the proxy; fall
+        # back to the price quoted at budget-check time so the session budget
+        # still accrues when the response omits cost fields.
+        charged = None
+        if isinstance(result, dict):
+            charged = result.get("price_per_call", result.get("cost"))
+        try:
+            self._session_spend += float(charged)
+        except (TypeError, ValueError):
+            if quoted_price is not None:
+                self._session_spend += quoted_price
+        return result
+
+
+@tool
+def registry_search(
+    query: str,
+    proxy_url: Optional[str] = None,
+    token: Optional[str] = None,
+) -> Dict[str, Any]:
+    """Search the connected tool registry by capability.
+
+    Free read. Returns catalogue matches for a natural-language capability query
+    such as "backlinks for a domain" or "company enrichment by email".
+
+    Args:
+        query: Capability to search for.
+        proxy_url: Registry base URL (defaults to TOOL_PROXY_URL env var).
+        token: Proxy token (defaults to TOOL_PROXY_TOKEN env var).
+
+    Returns:
+        Dict with catalogue matches, or ``{"error": ...}`` on failure.
+    """
+    return RegistryProxyTool(proxy_url=proxy_url, token=token).search(query)
+
+
+@tool
+def registry_describe(
+    tool_id: str,
+    proxy_url: Optional[str] = None,
+    token: Optional[str] = None,
+) -> Dict[str, Any]:
+    """Describe one registry catalogue entry.
+
+    Free read. Returns the endpoint's parameters, price-per-call and an example
+    response so the model can decide whether and how to invoke it.
+
+    Args:
+        tool_id: Catalogue entry identifier (from ``registry_search``).
+        proxy_url: Registry base URL (defaults to TOOL_PROXY_URL env var).
+        token: Proxy token (defaults to TOOL_PROXY_TOKEN env var).
+
+    Returns:
+        Dict describing the entry, or ``{"error": ...}`` on failure.
+    """
+    return RegistryProxyTool(proxy_url=proxy_url, token=token).describe(tool_id)
+
+
+@tool
+def registry_call(
+    tool_id: str,
+    params: Optional[Dict[str, Any]] = None,
+    proxy_url: Optional[str] = None,
+    token: Optional[str] = None,
+    max_cost_per_call: Optional[float] = None,
+    max_session_spend: Optional[float] = None,
+) -> Dict[str, Any]:
+    """Invoke a registry endpoint through the proxy.
+
+    Paid call. The registry injects the upstream vendor credential server-side,
+    so no vendor keys are ever held by the agent. Optional spend guards deny +
+    report the call when a configured budget would be exceeded.
+
+    Args:
+        tool_id: Catalogue entry identifier (from ``registry_search``).
+        params: Parameters for the upstream endpoint.
+        proxy_url: Registry base URL (defaults to TOOL_PROXY_URL env var).
+        token: Proxy token (defaults to TOOL_PROXY_TOKEN env var).
+        max_cost_per_call: Deny the call if its price exceeds this value.
+        max_session_spend: Deny the call if it would push cumulative spend past this value.
+
+    Returns:
+        Dict with the endpoint response, or ``{"error": ...}`` on failure/denial.
+    """
+    return RegistryProxyTool(
+        proxy_url=proxy_url,
+        token=token,
+        max_cost_per_call=max_cost_per_call,
+        max_session_spend=max_session_spend,
+    ).call(tool_id, params)

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "praisonai-tools"
-version = "0.2.36"
+version = "0.2.37"
 description = "Extended tools for PraisonAI Agents"
 authors = [
     {name = "Mervin Praison"}
@@ -64,6 +64,9 @@ agentfolio = [
 n8n = [
     "httpx>=0.27.0",
 ]
+registry-proxy = [
+    "httpx>=0.24.0",
+]
 langextract = [
     "langextract>=0.1.0",
 ]

tests/test_registry_proxy.py

@@ -0,0 +1,387 @@
+"""Unit tests for the tool-registry proxy connector."""
+
+import os
+import pytest
+from unittest.mock import Mock, patch, MagicMock
+
+
+@pytest.fixture(autouse=True)
+def reset_session_spend():
+    """Isolate cumulative-spend state between tests."""
+    from praisonai_tools.registry_proxy import registry_proxy as mod
+
+    mod._SESSION_SPEND.clear()
+    yield
+    mod._SESSION_SPEND.clear()
+
+
+@pytest.fixture
+def mock_httpx():
+    """Mock httpx module with proper exception classes."""
+    mock_module = MagicMock()
+    mock_module.TimeoutException = type("TimeoutException", (Exception,), {})
+
+    class MockHTTPStatusError(Exception):
+        def __init__(self, message, request=None, response=None):
+            super().__init__(message)
+            self.request = request
+            self.response = response
+
+    mock_module.HTTPStatusError = MockHTTPStatusError
+    with patch.dict("sys.modules", {"httpx": mock_module}):
+        yield mock_module
+
+
+class TestConfiguration:
+    def test_disabled_when_unconfigured(self):
+        from praisonai_tools.registry_proxy import RegistryProxyTool
+
+        with patch.dict(os.environ, {}, clear=True):
+            tool = RegistryProxyTool()
+            result = tool.search("backlinks for a domain")
+            assert "not configured" in result["error"]
+
+    def test_reads_env_vars(self):
+        from praisonai_tools.registry_proxy import RegistryProxyTool
+
+        with patch.dict(os.environ, {
+            "TOOL_PROXY_URL": "https://registry.example.com/",
+            "TOOL_PROXY_TOKEN": "secret-token",
+        }):
+            tool = RegistryProxyTool()
+            assert tool.proxy_url == "https://registry.example.com"
+            assert tool.token == "secret-token"
+            assert tool._headers()["Authorization"] == "Bearer secret-token"
+
+    def test_explicit_args_override_env(self):
+        from praisonai_tools.registry_proxy import RegistryProxyTool
+
+        tool = RegistryProxyTool(proxy_url="https://self.host/", token="tok")
+        assert tool.proxy_url == "https://self.host"
+        assert tool.token == "tok"
+
+    def test_env_token_not_leaked_to_supplied_url(self):
+        """Security: an agent-supplied proxy_url must not receive the env token."""
+        from praisonai_tools.registry_proxy import RegistryProxyTool
+
+        with patch.dict(os.environ, {
+            "TOOL_PROXY_URL": "https://trusted.example.com",
+            "TOOL_PROXY_TOKEN": "secret-token",
+        }):
+            tool = RegistryProxyTool(proxy_url="https://attacker.example.com")
+            assert tool.proxy_url == "https://attacker.example.com"
+            assert tool.token is None
+            assert "Authorization" not in tool._headers()
+
+    def test_env_token_used_for_env_url(self):
+        from praisonai_tools.registry_proxy import RegistryProxyTool
+
+        with patch.dict(os.environ, {
+            "TOOL_PROXY_URL": "https://trusted.example.com",
+            "TOOL_PROXY_TOKEN": "secret-token",
+        }):
+            tool = RegistryProxyTool()
+            assert tool._headers()["Authorization"] == "Bearer secret-token"
+
+
+class TestSearch:
+    def test_search_success(self, mock_httpx):
+        from praisonai_tools.registry_proxy import RegistryProxyTool
+
+        response = Mock()
+        response.json.return_value = {"matches": [{"id": "seo.backlinks"}]}
+        response.raise_for_status.return_value = None
+        client = Mock()
+        client.get.return_value = response
+        mock_httpx.Client.return_value.__enter__.return_value = client
+
+        tool = RegistryProxyTool(proxy_url="https://r.example.com", token="t")
+        result = tool.search("backlinks for a domain")
+
+        assert result["matches"][0]["id"] == "seo.backlinks"
+        client.get.assert_called_once_with(
+            "https://r.example.com/catalog/search",
+            params={"q": "backlinks for a domain"},
+            headers={"Content-Type": "application/json", "Authorization": "Bearer t"},
+        )
+
+    def test_search_requires_query(self):
+        from praisonai_tools.registry_proxy import RegistryProxyTool
+
+        tool = RegistryProxyTool(proxy_url="https://r.example.com")
+        assert tool.search("")["error"] == "query is required"
+
+
+class TestDescribe:
+    def test_describe_success(self, mock_httpx):
+        from praisonai_tools.registry_proxy import RegistryProxyTool
+
+        response = Mock()
+        response.json.return_value = {"id": "seo.backlinks", "price_per_call": 0.01}
+        response.raise_for_status.return_value = None
+        client = Mock()
+        client.get.return_value = response
+        mock_httpx.Client.return_value.__enter__.return_value = client
+
+        tool = RegistryProxyTool(proxy_url="https://r.example.com")
+        result = tool.describe("seo.backlinks")
+
+        assert result["price_per_call"] == 0.01
+        client.get.assert_called_once_with(
+            "https://r.example.com/catalog/seo.backlinks",
+            headers={"Content-Type": "application/json"},
+        )
+
+
+class TestCall:
+    def test_call_success(self, mock_httpx):
+        from praisonai_tools.registry_proxy import RegistryProxyTool
+
+        response = Mock()
+        response.json.return_value = {"data": {"backlinks": 42}}
+        response.raise_for_status.return_value = None
+        client = Mock()
+        client.post.return_value = response
+        mock_httpx.Client.return_value.__enter__.return_value = client
+
+        tool = RegistryProxyTool(proxy_url="https://r.example.com", token="t")
+        result = tool.call("seo.backlinks", {"domain": "example.com"})
+
+        assert result["data"]["backlinks"] == 42
+        client.post.assert_called_once_with(
+            "https://r.example.com/call/seo.backlinks",
+            json={"domain": "example.com"},
+            headers={"Content-Type": "application/json", "Authorization": "Bearer t"},
+        )
+
+    def test_call_requires_tool_id(self):
+        from praisonai_tools.registry_proxy import RegistryProxyTool
+
+        tool = RegistryProxyTool(proxy_url="https://r.example.com")
+        assert tool.call("")["error"] == "tool_id is required"
+
+
+class TestErrorTaxonomy:
+    def test_auth_error(self, mock_httpx):
+        from praisonai_tools.registry_proxy import RegistryProxyTool
+
+        resp = Mock()
+        resp.status_code = 401
+        resp.text = "Unauthorized"
+        client = Mock()
+        client.get.side_effect = mock_httpx.HTTPStatusError("401", response=resp)
+        mock_httpx.Client.return_value.__enter__.return_value = client
+
+        tool = RegistryProxyTool(proxy_url="https://r.example.com", token="bad")
+        result = tool.search("x")
+        assert "Authentication failed" in result["error"]
+
+    def test_insufficient_balance_error(self, mock_httpx):
+        from praisonai_tools.registry_proxy import RegistryProxyTool
+
+        resp = Mock()
+        resp.status_code = 402
+        resp.text = "Payment Required"
+        client = Mock()
+        client.post.side_effect = mock_httpx.HTTPStatusError("402", response=resp)
+        mock_httpx.Client.return_value.__enter__.return_value = client
+
+        tool = RegistryProxyTool(proxy_url="https://r.example.com")
+        result = tool.call("seo.backlinks", {})
+        assert "Insufficient balance" in result["error"]
+
+    def test_upstream_4xx_passthrough(self, mock_httpx):
+        from praisonai_tools.registry_proxy import RegistryProxyTool
+
+        resp = Mock()
+        resp.status_code = 404
+        resp.text = "Not Found"
+        client = Mock()
+        client.post.side_effect = mock_httpx.HTTPStatusError("404", response=resp)
+        mock_httpx.Client.return_value.__enter__.return_value = client
+
+        tool = RegistryProxyTool(proxy_url="https://r.example.com")
+        result = tool.call("seo.backlinks", {})
+        assert "HTTP 404" in result["error"]
+
+    def test_timeout_error(self, mock_httpx):
+        from praisonai_tools.registry_proxy import RegistryProxyTool
+
+        client = Mock()
+        client.get.side_effect = mock_httpx.TimeoutException("timed out")
+        mock_httpx.Client.return_value.__enter__.return_value = client
+
+        tool = RegistryProxyTool(proxy_url="https://r.example.com", timeout=5.0)
+        result = tool.search("x")
+        assert "timed out after 5.0" in result["error"]
+
+    def test_missing_httpx(self):
+        from praisonai_tools.registry_proxy import RegistryProxyTool
+
+        tool = RegistryProxyTool(proxy_url="https://r.example.com")
+        with patch.dict("sys.modules", {"httpx": None}):
+            with patch("builtins.__import__", side_effect=ImportError("no httpx")):
+                result = tool.search("x")
+                assert "httpx not installed" in result["error"]
+                assert "praisonai-tools[registry-proxy]" in result["error"]
+
+
+class TestSpendGuards:
+    def test_max_cost_per_call_denies(self, mock_httpx):
+        from praisonai_tools.registry_proxy import RegistryProxyTool
+
+        describe_resp = Mock()
+        describe_resp.json.return_value = {"id": "seo.backlinks", "price_per_call": 0.50}
+        describe_resp.raise_for_status.return_value = None
+        client = Mock()
+        client.get.return_value = describe_resp
+        mock_httpx.Client.return_value.__enter__.return_value = client
+
+        tool = RegistryProxyTool(proxy_url="https://r.example.com", max_cost_per_call=0.10)
+        result = tool.call("seo.backlinks", {})
+        assert "exceeds max_cost_per_call" in result["error"]
+        client.post.assert_not_called()
+
+    def test_max_session_spend_denies(self, mock_httpx):
+        from praisonai_tools.registry_proxy import RegistryProxyTool
+
+        describe_resp = Mock()
+        describe_resp.json.return_value = {"price_per_call": 0.60}
+        describe_resp.raise_for_status.return_value = None
+        client = Mock()
+        client.get.return_value = describe_resp
+        mock_httpx.Client.return_value.__enter__.return_value = client
+
+        tool = RegistryProxyTool(proxy_url="https://r.example.com", max_session_spend=1.0)
+        tool._session_spend = 0.50
+        result = tool.call("seo.backlinks", {})
+        assert "max_session_spend" in result["error"]
+        client.post.assert_not_called()
+
+    def test_within_budget_allows_and_tracks_spend(self, mock_httpx):
+        from praisonai_tools.registry_proxy import RegistryProxyTool
+
+        describe_resp = Mock()
+        describe_resp.json.return_value = {"price_per_call": 0.05}
+        describe_resp.raise_for_status.return_value = None
+        call_resp = Mock()
+        call_resp.json.return_value = {"data": "ok", "price_per_call": 0.05}
+        call_resp.raise_for_status.return_value = None
+        client = Mock()
+        client.get.return_value = describe_resp
+        client.post.return_value = call_resp
+        mock_httpx.Client.return_value.__enter__.return_value = client
+
+        tool = RegistryProxyTool(proxy_url="https://r.example.com", max_session_spend=1.0)
+        result = tool.call("seo.backlinks", {})
+        assert result["data"] == "ok"
+        assert tool._session_spend == pytest.approx(0.05)
+
+    def test_missing_price_fails_closed(self, mock_httpx):
+        from praisonai_tools.registry_proxy import RegistryProxyTool
+
+        describe_resp = Mock()
+        describe_resp.json.return_value = {"id": "seo.backlinks"}
+        describe_resp.raise_for_status.return_value = None
+        client = Mock()
+        client.get.return_value = describe_resp
+        mock_httpx.Client.return_value.__enter__.return_value = client
+
+        tool = RegistryProxyTool(proxy_url="https://r.example.com", max_cost_per_call=0.10)
+        result = tool.call("seo.backlinks", {})
+        assert "cannot enforce" in result["error"]
+        client.post.assert_not_called()
+
+    def test_session_spend_persists_across_registry_call(self, mock_httpx):
+        """max_session_spend must accumulate across separate registry_call calls."""
+        from praisonai_tools.registry_proxy import registry_call
+
+        describe_resp = Mock()
+        describe_resp.json.return_value = {"price_per_call": 0.60}
+        describe_resp.raise_for_status.return_value = None
+        call_resp = Mock()
+        call_resp.json.return_value = {"data": "ok", "price_per_call": 0.60}
+        call_resp.raise_for_status.return_value = None
+        client = Mock()
+        client.get.return_value = describe_resp
+        client.post.return_value = call_resp
+        mock_httpx.Client.return_value.__enter__.return_value = client
+
+        first = registry_call(
+            "seo.backlinks", {},
+            proxy_url="https://r.example.com", token="t",
+            max_session_spend=1.0,
+        )
+        assert first["data"] == "ok"
+
+        second = registry_call(
+            "seo.backlinks", {},
+            proxy_url="https://r.example.com", token="t",
+            max_session_spend=1.0,
+        )
+        assert "max_session_spend" in second["error"]
+
+
+class TestDecoratedFunctions:
+    @patch("praisonai_tools.registry_proxy.registry_proxy.RegistryProxyTool")
+    def test_registry_search_function(self, mock_cls):
+        from praisonai_tools.registry_proxy import registry_search
+
+        instance = Mock()
+        instance.search.return_value = {"matches": []}
+        mock_cls.return_value = instance
+
+        registry_search("q", proxy_url="https://r", token="t")
+        mock_cls.assert_called_once_with(proxy_url="https://r", token="t")
+        instance.search.assert_called_once_with("q")
+
+    @patch("praisonai_tools.registry_proxy.registry_proxy.RegistryProxyTool")
+    def test_registry_call_function_passes_guards(self, mock_cls):
+        from praisonai_tools.registry_proxy import registry_call
+
+        instance = Mock()
+        instance.call.return_value = {"data": "ok"}
+        mock_cls.return_value = instance
+
+        registry_call(
+            "seo.backlinks",
+            {"domain": "x"},
+            proxy_url="https://r",
+            token="t",
+            max_cost_per_call=0.1,
+            max_session_spend=1.0,
+        )
+        mock_cls.assert_called_once_with(
+            proxy_url="https://r",
+            token="t",
+            max_cost_per_call=0.1,
+            max_session_spend=1.0,
+        )
+        instance.call.assert_called_once_with("seo.backlinks", {"domain": "x"})
+
+
+def test_smoke_import():
+    from praisonai_tools.registry_proxy import (
+        RegistryProxyTool,
+        registry_search,
+        registry_describe,
+        registry_call,
+    )
+
+    assert callable(registry_search)
+    assert callable(registry_describe)
+    assert callable(registry_call)
+    assert RegistryProxyTool().name == "registry_proxy"
+
+
+@pytest.mark.integration
+def test_live_smoke():
+    """Gated live smoke test; skipped unless TOOL_PROXY_URL is configured."""
+    proxy_url = os.getenv("TOOL_PROXY_URL")
+    if not proxy_url:
+        pytest.skip("TOOL_PROXY_URL not set")
+    from praisonai_tools.registry_proxy import registry_search
+
+    result = registry_search("backlinks for a domain")
+    assert isinstance(result, dict)
+    assert "error" not in result

tests/test_registry_skills_compat.py

@@ -0,0 +1,25 @@
+"""Tier 0: pin that registry-installed skills (./.claude/skills/) are discoverable.
+
+Registries that expose ``skill install <name>`` install into ``./.claude/skills/``.
+PraisonAI's skill discovery already scans that directory, so registry-installed
+skills are discoverable with zero connector code. This test pins that contract.
+"""
+
+import os
+import pytest
+
+
+def test_claude_skills_dir_is_scanned(tmp_path, monkeypatch):
+    discovery = pytest.importorskip("praisonaiagents.skills.discovery")
+
+    skills_dir = tmp_path / ".claude" / "skills" / "my-registry-skill"
+    skills_dir.mkdir(parents=True)
+
+    monkeypatch.chdir(tmp_path)
+    dirs = discovery.get_default_skill_dirs()
+
+    claude_skills = tmp_path / ".claude" / "skills"
+    assert any(os.path.samefile(str(d), str(claude_skills)) for d in dirs), (
+        "PraisonAI must scan ./.claude/skills/ so registry-installed skills are "
+        "discoverable with zero connector code."
+    )