Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions src/smallestai/atoms/helpers/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,8 @@
from smallestai.atoms.helpers.call import Call, CallAnalytics
from smallestai.atoms.helpers.campaign import Campaign
from smallestai.atoms.helpers.kb import KB
from smallestai.atoms.helpers.secrets import Secrets
from smallestai.atoms.helpers.tools import Tools
from smallestai.atoms.helpers.versioning import (
BaseRevisionUnavailableError,
DraftConflictError,
Expand All @@ -29,6 +31,8 @@
"Call",
"Campaign",
"KB",
"Secrets",
"Tools",
"Page",
"as_page",
"require_id",
Expand Down
76 changes: 76 additions & 0 deletions src/smallestai/atoms/helpers/secrets.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,76 @@
"""
Secrets vault.

The org-level, write-only Secrets vault: store API keys and tokens encrypted and
reference them by name from a tool's ``auth`` block. Values are never returned —
``list`` and ``create`` surface the name only.

Usage:
from smallestai.atoms.helpers import Secrets

secrets = Secrets()
secrets.create(name="ORDER_API_TOKEN", value="sk_live_...")
secrets.list() # names only
secrets.delete(secret_id)
"""

import os
from typing import Any, Dict, Optional

import requests

DEFAULT_BASE_URL = "https://api.smallest.ai/atoms/v1"


class Secrets:
"""Manager for the org Secrets vault (``/secret``).

Standalone:
secrets = Secrets()
secrets.list()
"""

def __init__(
self,
base_url: Optional[str] = None,
api_key: Optional[str] = None,
):
"""
Args:
base_url: API base URL (default: api.smallest.ai/atoms/v1)
api_key: API key (default: SMALLEST_API_KEY env var)
"""
self.base_url = base_url or os.environ.get("SMALLEST_BASE_URL", DEFAULT_BASE_URL)
self.api_key = api_key or os.environ.get("SMALLEST_API_KEY", "")

def _get_headers(self) -> Dict[str, str]:
return {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
}

def list(self) -> Dict[str, Any]:
"""List the org's secret names (values are never returned)."""
url = f"{self.base_url}/secret"
response = requests.get(url, headers=self._get_headers())
response.raise_for_status()
return response.json() # type: ignore[no-any-return]

def create(self, name: str, value: str) -> Dict[str, Any]:
"""Create a secret. The value is encrypted at rest; the response holds the name only.

Args:
name: secret name (letters, numbers, underscores) — referenced from a tool's ``auth``.
value: the secret value (write-only).
"""
url = f"{self.base_url}/secret"
response = requests.post(url, headers=self._get_headers(), json={"name": name, "value": value})
response.raise_for_status()
return response.json() # type: ignore[no-any-return]

def delete(self, secret_id: str) -> Dict[str, Any]:
"""Delete a secret by its id."""
url = f"{self.base_url}/secret/{secret_id}"
response = requests.delete(url, headers=self._get_headers())
response.raise_for_status()
return response.json() # type: ignore[no-any-return]
103 changes: 103 additions & 0 deletions src/smallestai/atoms/helpers/tools.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,103 @@
"""
Reusable Tools library.

The org-level Tools registry: create a tool once and reference it from any number
of agents by its ``toolId``. Two types are registry tools — ``api_call`` and
``client_tool``. System tools (transfer, end-call, knowledge-base search) are
configured per-agent, not here.

Usage:
from smallestai.atoms.helpers import Tools

tools = Tools()
created = tools.create({
"type": "api_call",
"name": "get_order_status",
"description": "Look up an order by id.",
"method": "GET",
"url": "https://api.example.com/orders/{{order_id}}",
"llmParameters": [
{"name": "order_id", "type": "text", "description": "The order id", "required": True}
],
})
tool_id = created["data"]["toolId"]
tools.list()
tools.update(tool_id, {...})
tools.duplicate(tool_id)
tools.delete(tool_id)
"""

import os
from typing import Any, Dict, Optional

import requests

DEFAULT_BASE_URL = "https://api.smallest.ai/atoms/v1"


class Tools:
"""Manager for the org Tools library (``/tool``).

Standalone:
tools = Tools()
tools.list()
"""

def __init__(
self,
base_url: Optional[str] = None,
api_key: Optional[str] = None,
):
"""
Args:
base_url: API base URL (default: api.smallest.ai/atoms/v1)
api_key: API key (default: SMALLEST_API_KEY env var)
"""
self.base_url = base_url or os.environ.get("SMALLEST_BASE_URL", DEFAULT_BASE_URL)
self.api_key = api_key or os.environ.get("SMALLEST_API_KEY", "")

def _get_headers(self) -> Dict[str, str]:
return {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
}

def list(self) -> Dict[str, Any]:
"""List the org's tools."""
url = f"{self.base_url}/tool"
response = requests.get(url, headers=self._get_headers())
response.raise_for_status()
return response.json() # type: ignore[no-any-return]

def create(self, definition: Dict[str, Any]) -> Dict[str, Any]:
"""Create a tool.

Args:
definition: the tool body (``type`` ``api_call`` or ``client_tool``,
plus ``name``, ``description``, and the fields for that type).
"""
url = f"{self.base_url}/tool"
response = requests.post(url, headers=self._get_headers(), json={"definition": definition})
response.raise_for_status()
return response.json() # type: ignore[no-any-return]

def update(self, tool_id: str, definition: Dict[str, Any]) -> Dict[str, Any]:
"""Update a tool. Propagates to every agent that references it."""
url = f"{self.base_url}/tool/{tool_id}"
response = requests.patch(url, headers=self._get_headers(), json={"definition": definition})
response.raise_for_status()
return response.json() # type: ignore[no-any-return]

def duplicate(self, tool_id: str) -> Dict[str, Any]:
"""Duplicate a tool into a new library tool with a new ``toolId``."""
url = f"{self.base_url}/tool/{tool_id}/duplicate"
response = requests.post(url, headers=self._get_headers())
response.raise_for_status()
return response.json() # type: ignore[no-any-return]

def delete(self, tool_id: str) -> Dict[str, Any]:
"""Delete a tool. Blocked (400) while any agent still references it."""
url = f"{self.base_url}/tool/{tool_id}"
response = requests.delete(url, headers=self._get_headers())
response.raise_for_status()
return response.json() # type: ignore[no-any-return]
32 changes: 31 additions & 1 deletion tests/custom/test_helpers_rest.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,9 @@
import smallestai.atoms.helpers.audience as audience_mod
import smallestai.atoms.helpers.campaign as campaign_mod
import smallestai.atoms.helpers.kb as kb_mod
from smallestai.atoms.helpers import KB, Audience, Campaign, Page
import smallestai.atoms.helpers.secrets as secrets_mod
import smallestai.atoms.helpers.tools as tools_mod
from smallestai.atoms.helpers import KB, Audience, Campaign, Page, Secrets, Tools


class _FakeResp:
Expand Down Expand Up @@ -77,3 +79,31 @@ def test_campaign_list_builds_url_with_auth(monkeypatch):
method, url, headers = fake.last
assert method == "get" and url.startswith("https://api.example/atoms/v1")
assert headers.get("Authorization") == "Bearer sk_test"


def test_tools_crud_build_urls_with_auth(monkeypatch):
fake = _patch(monkeypatch, tools_mod)
tools = Tools(base_url="https://api.example/atoms/v1", api_key="sk_test")
tools.list()
assert fake.last[0] == "get" and fake.last[1].endswith("/tool")
tools.create({"type": "api_call", "name": "t", "description": "d"})
assert fake.last[0] == "post" and fake.last[1].endswith("/tool")
tools.update("tool_abc", {"type": "api_call", "name": "t", "description": "d"})
assert fake.last[0] == "patch" and fake.last[1].endswith("/tool/tool_abc")
tools.duplicate("tool_abc")
assert fake.last[0] == "post" and fake.last[1].endswith("/tool/tool_abc/duplicate")
tools.delete("tool_abc")
assert fake.last[0] == "delete" and fake.last[1].endswith("/tool/tool_abc")
assert fake.last[2].get("Authorization") == "Bearer sk_test"


def test_secrets_crud_build_urls_with_auth(monkeypatch):
fake = _patch(monkeypatch, secrets_mod)
secrets = Secrets(base_url="https://api.example/atoms/v1", api_key="sk_test")
secrets.list()
assert fake.last[0] == "get" and fake.last[1].endswith("/secret")
secrets.create(name="MY_SECRET", value="v")
assert fake.last[0] == "post" and fake.last[1].endswith("/secret")
secrets.delete("secret_id_1")
assert fake.last[0] == "delete" and fake.last[1].endswith("/secret/secret_id_1")
assert fake.last[2].get("Authorization") == "Bearer sk_test"
Loading