From 2e7ae775c3fd8e7e9c74528cd2b3b81001ab9387 Mon Sep 17 00:00:00 2001 From: Calvin Grunewald Date: Fri, 12 Jun 2026 14:06:35 -0700 Subject: [PATCH 1/3] add python sdk examples --- .github/workflows/ci.yml | 2 +- README.md | 3 + examples/create_agent_cli/README.md | 41 ++ examples/create_agent_cli/main.py | 113 +++++ examples/org_system_user_token/README.md | 156 +++++++ examples/org_system_user_token/main.py | 44 ++ examples/thread_chat_tui/README.md | 38 ++ examples/thread_chat_tui/main.py | 424 +++++++++++++++++++ tests/examples/test_create_agent_cli.py | 116 +++++ tests/examples/test_org_system_user_token.py | 54 +++ tests/examples/test_thread_chat_tui.py | 297 +++++++++++++ 11 files changed, 1287 insertions(+), 1 deletion(-) create mode 100644 examples/create_agent_cli/README.md create mode 100644 examples/create_agent_cli/main.py create mode 100644 examples/org_system_user_token/README.md create mode 100644 examples/org_system_user_token/main.py create mode 100644 examples/thread_chat_tui/README.md create mode 100644 examples/thread_chat_tui/main.py create mode 100644 tests/examples/test_create_agent_cli.py create mode 100644 tests/examples/test_org_system_user_token.py create mode 100644 tests/examples/test_thread_chat_tui.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index a1109fe..1ec7aa0 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -57,7 +57,7 @@ jobs: run: uv run ruff format --check - name: Unit tests - run: uv run pytest tests/test_http_client.py src/archastro/phx_channel/tests/test_unit.py + run: uv run pytest tests/test_http_client.py src/archastro/phx_channel/tests/test_unit.py tests/examples - name: Harness-client integration tests run: uv run pytest tests/harness diff --git a/README.md b/README.md index df1fd60..a0ac45b 100644 --- a/README.md +++ b/README.md @@ -50,6 +50,9 @@ uv sync --locked --all-extras # Unit tests only (no external services needed) uv run pytest tests/test_http_client.py src/archastro/phx_channel/tests/test_unit.py +# Example smoke/unit tests +uv run pytest tests/examples + # REST contract tests (spawns Prism mock server) uv run pytest tests/contract diff --git a/examples/create_agent_cli/README.md b/examples/create_agent_cli/README.md new file mode 100644 index 0000000..3f02fc7 --- /dev/null +++ b/examples/create_agent_cli/README.md @@ -0,0 +1,41 @@ +# Create Agent CLI + +This example shows how to wrap the Python SDK in a small CLI for creating an +agent. + +The SDK call is intentionally direct: + +```python +with PlatformClient.with_token(api_key, token, base_url=base_url) as client: + agent = client.agents.create({...}) +``` + +## Run + +```bash +export ARCHASTRO_API_KEY=pk_... +export ARCHASTRO_ACCESS_TOKEN=sat_... + +uv run python examples/create_agent_cli/main.py \ + --name "Demo Agent" \ + --identity "You are a concise assistant for onboarding users." +``` + +To create the agent under a specific org or team: + +```bash +uv run python examples/create_agent_cli/main.py \ + --name "Team Demo Agent" \ + --identity "You help the team answer support questions." \ + --org org_... \ + --team team_... +``` + +For local development or another environment: + +```bash +export ARCHASTRO_PLATFORM_BASE_URL=http://localhost:4000 +uv run python examples/create_agent_cli/main.py \ + --name "Local Demo Agent" \ + --identity "You are running from the Python SDK example." +``` diff --git a/examples/create_agent_cli/main.py b/examples/create_agent_cli/main.py new file mode 100644 index 0000000..88b8955 --- /dev/null +++ b/examples/create_agent_cli/main.py @@ -0,0 +1,113 @@ +# Copyright (c) 2026 ArchAstro Inc. All Rights Reserved. + +from __future__ import annotations + +import argparse +import json +import os +from typing import Any + +from pydantic import BaseModel + +from archastro.platform import PlatformClient + +DEFAULT_PLATFORM_BASE_URL = "https://platform.archastro.ai" + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Create a basic ArchAstro agent using the Python SDK." + ) + parser.add_argument("--name", required=True, help="Agent display name.") + parser.add_argument("--identity", required=True, help="Identity prompt for the agent.") + parser.add_argument("--model", help="Default model identifier for the agent.") + parser.add_argument("--org", help="Org id that should own the agent.") + parser.add_argument("--team", help="Team id that should own the agent.") + parser.add_argument("--user", help="User id that should own the agent.") + parser.add_argument("--lookup-key", help="Stable lookup key for idempotent external scripts.") + parser.add_argument("--template", help="Existing template id or lookup key to provision from.") + parser.add_argument("--originator", help="Free-form source label for the created agent.") + parser.add_argument( + "--metadata-json", + type=_json_object, + help='Optional metadata object, for example \'{"source":"python-sdk-example"}\'.', + ) + parser.add_argument( + "--base-url", + default=_env("ARCHASTRO_PLATFORM_BASE_URL", "ARCHASTRO_BASE_URL") + or DEFAULT_PLATFORM_BASE_URL, + help=( + "Platform base URL. Defaults to ARCHASTRO_PLATFORM_BASE_URL, " + "ARCHASTRO_BASE_URL, or production." + ), + ) + return parser.parse_args(argv) + + +def build_agent_input(args: argparse.Namespace) -> dict[str, object]: + fields = { + "name": args.name, + "identity": args.identity, + "model": args.model, + "org": args.org, + "team": args.team, + "user": args.user, + "lookup_key": args.lookup_key, + "template": args.template, + "originator": args.originator, + "metadata": args.metadata_json, + } + return {key: value for key, value in fields.items() if value is not None} + + +def create_agent(args: argparse.Namespace) -> dict[str, Any]: + api_key = _required_env("ARCHASTRO_API_KEY") + token = _required_env("ARCHASTRO_ACCESS_TOKEN") + + with PlatformClient.with_token(api_key, token, base_url=args.base_url) as client: + agent = client.agents.create(build_agent_input(args)) + return _plain(agent) + + +def _json_object(value: str) -> dict[str, object]: + try: + parsed = json.loads(value) + except json.JSONDecodeError as exc: + raise argparse.ArgumentTypeError(str(exc)) from exc + if not isinstance(parsed, dict): + raise argparse.ArgumentTypeError("--metadata-json must be a JSON object") + return parsed + + +def _env(*names: str) -> str | None: + for name in names: + value = os.environ.get(name) + if value: + return value + return None + + +def _required_env(name: str) -> str: + value = _env(name) + if not value: + raise SystemExit(f"Set {name} before running this example.") + return value + + +def _plain(value: Any) -> dict[str, Any]: + if isinstance(value, BaseModel): + return value.model_dump(mode="json", exclude_none=True) + if isinstance(value, dict): + return value + if hasattr(value, "model_dump"): + return value.model_dump(mode="json", exclude_none=True) + return {"result": value} + + +def main() -> None: + agent = create_agent(parse_args()) + print(json.dumps(agent, indent=2, sort_keys=True)) + + +if __name__ == "__main__": + main() diff --git a/examples/org_system_user_token/README.md b/examples/org_system_user_token/README.md new file mode 100644 index 0000000..59d8ad5 --- /dev/null +++ b/examples/org_system_user_token/README.md @@ -0,0 +1,156 @@ +# Org System-User Token + +Use this pattern when a Python process should act as an ArchAgents org-owned +service user: a bot, worker, cron job, ingestion process, or integration that +belongs to the org rather than to a human session. + +The Python process only needs one secret: + +```bash +export ARCHASTRO_ACCESS_TOKEN=sat_... +``` + +The backend derives the app and org context from that token, so this example +does not require `ARCHASTRO_API_KEY`. + +The clients default to the production API gateway, +`https://platform.archastro.ai`. + +For local development or another environment, override the base URL: + +```bash +export ARCHASTRO_PLATFORM_BASE_URL=http://localhost:4000 +``` + +## Use the Sync Client + +Use `PlatformClient` for scripts, CLIs, cron jobs, and small workers that do not +already run an event loop. + +```python +import os + +from archastro.platform import PlatformClient + +base_url = os.environ.get("ARCHASTRO_PLATFORM_BASE_URL", "https://platform.archastro.ai") +token = os.environ["ARCHASTRO_ACCESS_TOKEN"] + +with PlatformClient(base_url=base_url, access_token=token) as client: + user = client.users.me() + orgs = client.users.orgs(user["id"]) + +print(user["id"], user.get("is_system_user"), orgs) +``` + +Run the complete sync example: + +```bash +uv run python examples/org_system_user_token/main.py +``` + +## Use the Async Client + +Use `AsyncPlatformClient` inside async web apps, async workers, or code that also +uses websocket channels. + +```python +import asyncio +import os + +from archastro.platform import AsyncPlatformClient + + +async def main() -> None: + base_url = os.environ.get("ARCHASTRO_PLATFORM_BASE_URL", "https://platform.archastro.ai") + token = os.environ["ARCHASTRO_ACCESS_TOKEN"] + + async with AsyncPlatformClient(base_url=base_url, access_token=token) as client: + user = await client.users.me() + orgs = await client.users.orgs(user["id"]) + + print(user["id"], user.get("is_system_user"), orgs) + + +asyncio.run(main()) +``` + +## If You Are an ArchAgents User + +If you are already using ArchAgents and can log in as an org admin, create the +system user and token with `archagent`. + +```bash +archagent auth login +export ARCHASTRO_ORG_ID=org_... +``` + +Create an org-scoped system user. `member` is the safest default role for most +bots and integrations. + +```bash +export ARCHASTRO_SYSTEM_USER_ID="$( + archagent --json create user \ + --system-user \ + --name "ArchAgents Python SDK Bot" \ + --org "$ARCHASTRO_ORG_ID" \ + --org-role member | + jq -r '.id' +)" +``` + +Create a token for that system user. The raw token is shown once, so put it in +your secret manager immediately. + +```bash +export ARCHASTRO_ACCESS_TOKEN="$( + archagent --json create usertoken \ + --user "$ARCHASTRO_SYSTEM_USER_ID" \ + --name "python-sdk-service" | + jq -r '.token' +)" +``` + +Now run your Python service with `ARCHASTRO_ACCESS_TOKEN` in its environment. +Only set `ARCHASTRO_PLATFORM_BASE_URL` when targeting local development, +staging, or another non-production environment. + +## If You Are a Platform Developer + +Use `archastro` when you need to bootstrap the system user from a developer app +context instead of an ArchAgents org session. Run from an initialized ArchAstro +project, or pass `--app ` to each command. + +```bash +archastro auth login +export ARCHASTRO_ORG_ID=org_... + +export ARCHASTRO_SYSTEM_USER_ID="$( + archastro --json create user \ + --system-user \ + --name "ArchAgents Python SDK Bot" \ + --org "$ARCHASTRO_ORG_ID" \ + --org-role member | + jq -r '.id' +)" + +export ARCHASTRO_ACCESS_TOKEN="$( + archastro --json create usertoken \ + --user "$ARCHASTRO_SYSTEM_USER_ID" \ + --name "python-sdk-service" | + jq -r '.token' +)" +``` + +## Rotate or Revoke the Token + +List active tokens: + +```bash +archagent list usertokens --user "$ARCHASTRO_SYSTEM_USER_ID" +``` + +Revoke one token: + +```bash +archagent revoke usertoken sat_... --user "$ARCHASTRO_SYSTEM_USER_ID" +``` diff --git a/examples/org_system_user_token/main.py b/examples/org_system_user_token/main.py new file mode 100644 index 0000000..f460819 --- /dev/null +++ b/examples/org_system_user_token/main.py @@ -0,0 +1,44 @@ +# Copyright (c) 2026 ArchAstro Inc. All Rights Reserved. + +from __future__ import annotations + +import os + +from archastro.platform import PlatformClient + +DEFAULT_PLATFORM_BASE_URL = "https://platform.archastro.ai" + + +def main() -> None: + token = os.environ.get("ARCHASTRO_ACCESS_TOKEN") + if not token: + raise SystemExit( + "Set ARCHASTRO_ACCESS_TOKEN to a system-user token created by archagent or archastro." + ) + + base_url = ( + _env("ARCHASTRO_PLATFORM_BASE_URL", "ARCHASTRO_BASE_URL") or DEFAULT_PLATFORM_BASE_URL + ) + with PlatformClient(base_url=base_url, access_token=token) as client: + user = client.users.me() + print(f"Authenticated as {user['id']}") + + is_system_user = user.get("is_system_user") + if is_system_user is not None: + print(f"System user: {is_system_user}") + if user.get("org_role"): + print(f"Org role: {user['org_role']}") + if user.get("sandbox_id"): + print(f"Sandbox: {user['sandbox_id']}") + + +def _env(*names: str) -> str | None: + for name in names: + value = os.environ.get(name) + if value: + return value + return None + + +if __name__ == "__main__": + main() diff --git a/examples/thread_chat_tui/README.md b/examples/thread_chat_tui/README.md new file mode 100644 index 0000000..38509ff --- /dev/null +++ b/examples/thread_chat_tui/README.md @@ -0,0 +1,38 @@ +# Thread Chat TUI + +This example shows the websocket part of the Python SDK with a small +terminal chat UI around it. + +The important SDK flow is visible in `main.py`: + +1. Create an `AsyncPlatformClient` with the publishable API key and access token. +2. Open a websocket with `await client.open_socket()`. +3. Join a thread with `ApiChatChannel.join_user_thread(...)` or + `ApiChatChannel.join_team_thread(...)`. +4. Wrap the generated channel in `ThreadChatSession`. +5. Pass the session into `ThreadChatTui`. +6. Keep websocket operations in `ThreadChatSession` and terminal rendering in + `ThreadChatTui`. + +## Run + +```bash +export ARCHASTRO_API_KEY=pk_... +export ARCHASTRO_ACCESS_TOKEN=sat_... +uv run python examples/thread_chat_tui/main.py thr_... +``` + +For a team-scoped thread: + +```bash +uv run python examples/thread_chat_tui/main.py thr_... --team team_... +``` + +For local development or another environment: + +```bash +export ARCHASTRO_PLATFORM_BASE_URL=http://localhost:4000 +uv run python examples/thread_chat_tui/main.py thr_... +``` + +Inside the UI, press Enter to send and Ctrl-D or Ctrl-C to exit. diff --git a/examples/thread_chat_tui/main.py b/examples/thread_chat_tui/main.py new file mode 100644 index 0000000..241c06c --- /dev/null +++ b/examples/thread_chat_tui/main.py @@ -0,0 +1,424 @@ +# Copyright (c) 2026 ArchAstro Inc. All Rights Reserved. + +from __future__ import annotations + +import argparse +import asyncio +import curses +import os +import textwrap +import uuid +from collections.abc import Callable +from contextlib import suppress +from dataclasses import dataclass, replace +from typing import Any + +from archastro.platform import AsyncPlatformClient +from archastro.platform.channels.api_chat_channel import ApiChatChannel + +DEFAULT_PLATFORM_BASE_URL = "https://platform.archastro.ai" + + +@dataclass(frozen=True) +class ChatMessage: + id: str + author: str + content: str + created_at: str | None = None + idempotency_key: str | None = None + state: str | None = None + + +class ThreadChatSession: + def __init__(self, channel: ApiChatChannel): + self._channel = channel + + def on_message_added(self, callback: Callable[[dict[str, object]], None]) -> Callable[[], None]: + return self._channel.on_message_added(callback) + + async def load_history(self) -> list[ChatMessage]: + reply = await self._channel.api_chat_list_messages({}) + return messages_from_reply(reply) + + async def send_message(self, content: str, *, idempotency_key: str) -> None: + reply = await self._channel.api_chat_post_simple_message( + {"content": content, "idempotency_key": idempotency_key} + ) + if reply.get("status") == "ok": + return + raise RuntimeError(f"Send rejected: {reply.get('response', reply)}") + + async def close(self) -> None: + await self._channel.leave() + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Chat in an existing ArchAstro thread over the websocket SDK." + ) + parser.add_argument( + "thread_id", metavar="THREAD_ID", help="Thread id to join, for example thr_..." + ) + parser.add_argument( + "--team", + dest="team_id", + help="Join the thread as a team-scoped chat channel instead of a user-scoped channel.", + ) + parser.add_argument( + "--base-url", + default=_env("ARCHASTRO_PLATFORM_BASE_URL", "ARCHASTRO_BASE_URL") + or DEFAULT_PLATFORM_BASE_URL, + help=( + "Platform base URL. Defaults to ARCHASTRO_PLATFORM_BASE_URL, " + "ARCHASTRO_BASE_URL, or production." + ), + ) + parser.add_argument( + "--ws-url", + default=_env("ARCHASTRO_PLATFORM_WS_URL"), + help=( + "Optional websocket URL override. By default AsyncPlatformClient derives " + "the /socket/api/websocket URL from --base-url." + ), + ) + parser.add_argument( + "--api-key", + default=_env("ARCHASTRO_API_KEY"), + help="Publishable app API key. Defaults to ARCHASTRO_API_KEY.", + ) + parser.add_argument( + "--access-token", + default=_env("ARCHASTRO_ACCESS_TOKEN"), + help="User access token. Defaults to ARCHASTRO_ACCESS_TOKEN.", + ) + parser.add_argument( + "--limit", + type=int, + default=25, + help="Initial message limit requested when joining the channel.", + ) + return parser.parse_args(argv) + + +def message_from_payload(payload: dict[str, object]) -> ChatMessage: + raw_message = payload.get("message", payload) + if not isinstance(raw_message, dict): + raw_message = {} + + return ChatMessage( + id=str(raw_message.get("id") or f"message:{uuid.uuid4()}"), + author=author_for(raw_message), + content=str(raw_message.get("content") or ""), + created_at=_string_or_none(raw_message.get("created_at")), + idempotency_key=_string_or_none(raw_message.get("idempotency_key")), + ) + + +def author_for(message: dict[str, object]) -> str: + actors = message.get("actors") + if isinstance(actors, list) and actors: + actor = actors[0] + if isinstance(actor, dict): + for key in ("name", "alias", "id"): + value = actor.get(key) + if isinstance(value, str) and value: + return value + + user = message.get("user") + if isinstance(user, dict): + for key in ("name", "alias", "id"): + value = user.get(key) + if isinstance(value, str) and value: + return value + if isinstance(user, str) and user: + return user + + agent = message.get("agent") + if isinstance(agent, str) and agent: + return agent + + return "unknown" + + +def render_message(message: ChatMessage, width: int) -> list[str]: + state = f" [{message.state}]" if message.state else "" + content = f"{message.content}{state}".strip() + prefix = f"{message.author}: " + first_width = max(8, width - len(prefix)) + later_width = max(8, width - 2) + wrapped = textwrap.wrap(content, width=first_width) or [""] + + lines = [f"{prefix}{wrapped[0]}"] + for chunk in textwrap.wrap(" ".join(wrapped[1:]), width=later_width): + lines.append(f" {chunk}") + return [line[:width] for line in lines] + + +def messages_from_reply(reply: dict[str, object]) -> list[ChatMessage]: + payload = reply.get("response", reply) + messages = _messages_list(payload) + return [message_from_payload({"message": message}) for message in messages] + + +def _messages_list(payload: object) -> list[dict[str, object]]: + if isinstance(payload, list): + return [item for item in payload if isinstance(item, dict)] + if not isinstance(payload, dict): + return [] + + for key in ("messages", "data"): + value = payload.get(key) + if isinstance(value, list): + return [item for item in value if isinstance(item, dict)] + if isinstance(value, dict): + nested = _messages_list(value) + if nested: + return nested + return [] + + +def _string_or_none(value: object) -> str | None: + return value if isinstance(value, str) and value else None + + +class ThreadChatTui: + def __init__(self, session: ThreadChatSession, *, thread_id: str, team_id: str | None): + self._session = session + self._thread_id = thread_id + self._team_id = team_id + self._messages: list[ChatMessage] = [] + self._seen_message_ids: set[str] = set() + self._draft = "" + self._status = "Connected. Enter sends, Ctrl-D or Ctrl-C exits." + self._running = True + self._dirty = True + self._send_tasks: set[asyncio.Task[None]] = set() + + def add_history(self, messages: list[ChatMessage]) -> None: + for message in messages: + self._append_or_replace(message) + self._dirty = True + + def add_message_payload(self, payload: dict[str, object]) -> None: + self._append_or_replace(message_from_payload(payload)) + self._dirty = True + + def set_status(self, status: str) -> None: + self._status = status + self._dirty = True + + async def run(self) -> None: + screen = None + + try: + screen = curses.initscr() + curses.noecho() + curses.cbreak() + with suppress(curses.error): + curses.curs_set(1) + screen.keypad(True) + screen.nodelay(True) + + while self._running: + self._read_keys(screen) + if self._dirty: + self._draw(screen) + await asyncio.sleep(0.03) + finally: + for task in self._send_tasks: + task.cancel() + if self._send_tasks: + await asyncio.gather(*self._send_tasks, return_exceptions=True) + if screen is not None: + with suppress(curses.error): + screen.nodelay(False) + with suppress(curses.error): + screen.keypad(False) + with suppress(curses.error): + curses.nocbreak() + with suppress(curses.error): + curses.echo() + with suppress(curses.error): + curses.endwin() + + def _read_keys(self, screen: Any) -> None: + while True: + try: + key = screen.get_wch() + except curses.error: + return + + if key in ("\x03", "\x04"): + self._running = False + self._dirty = True + return + if key in ("\n", "\r"): + self._submit_draft() + continue + if key in ("\b", "\x7f") or key == curses.KEY_BACKSPACE: + self._draft = self._draft[:-1] + self._dirty = True + continue + if key == curses.KEY_RESIZE: + self._dirty = True + continue + if isinstance(key, str) and key.isprintable(): + self._draft += key + self._dirty = True + + def _submit_draft(self) -> None: + content = self._draft.strip() + self._draft = "" + self._dirty = True + if not content: + return + task = asyncio.create_task(self._post_message(content)) + self._send_tasks.add(task) + task.add_done_callback(self._send_tasks.discard) + + async def _post_message(self, content: str) -> None: + idempotency_key = str(uuid.uuid4()) + local = ChatMessage( + id=f"local:{idempotency_key}", + author="you", + content=content, + idempotency_key=idempotency_key, + state="sending", + ) + self._messages.append(local) + self._dirty = True + + try: + await self._session.send_message(content, idempotency_key=idempotency_key) + except Exception as exc: + self._replace_pending(idempotency_key, replace(local, state="failed")) + self._status = f"Send failed: {exc}" + self._dirty = True + return + + self._replace_pending(idempotency_key, replace(local, state="sent")) + self._status = "Message sent." + self._dirty = True + + def _append_or_replace(self, message: ChatMessage) -> None: + if message.idempotency_key and self._replace_pending(message.idempotency_key, message): + self._seen_message_ids.add(message.id) + return + if message.id in self._seen_message_ids: + return + self._messages.append(message) + self._seen_message_ids.add(message.id) + + def _replace_pending(self, idempotency_key: str, replacement: ChatMessage) -> bool: + for index, message in enumerate(self._messages): + if message.idempotency_key == idempotency_key: + self._messages[index] = replacement + return True + return False + + def _draw(self, screen: Any) -> None: + height, width = screen.getmaxyx() + width = max(20, width) + screen.erase() + + if height < 4: + screen.addnstr(0, 0, "Make the terminal taller to chat.".ljust(width), width) + screen.refresh() + self._dirty = False + return + + scope = f"team {self._team_id}" if self._team_id else "user" + header = f" ArchAstro chat | {scope} | {self._thread_id} " + screen.addnstr(0, 0, header.ljust(width), width, curses.A_REVERSE) + screen.addnstr(1, 0, self._status.ljust(width), width) + + body_top = 2 + body_bottom = max(body_top, height - 2) + body_height = max(1, body_bottom - body_top) + lines = self._render_transcript(width) + visible_lines = lines[-body_height:] + for offset, line in enumerate(visible_lines): + screen.addnstr(body_top + offset, 0, line.ljust(width), width) + + prompt = f"> {self._draft}" + prompt_width = max(1, width - 1) + if len(prompt) > prompt_width: + prompt = "> " + self._draft[-(prompt_width - 2) :] + screen.addnstr(height - 1, 0, prompt.ljust(prompt_width), prompt_width, curses.A_BOLD) + screen.move(height - 1, min(len(prompt), prompt_width - 1)) + screen.refresh() + self._dirty = False + + def _render_transcript(self, width: int) -> list[str]: + if not self._messages: + return ["No messages loaded yet. Type a message and press Enter."] + + lines: list[str] = [] + for message in self._messages: + lines.extend(render_message(message, width)) + lines.append("") + return lines[:-1] + + +async def chat(args: argparse.Namespace) -> None: + if not args.api_key: + raise SystemExit("Set ARCHASTRO_API_KEY before running this example.") + if not args.access_token: + raise SystemExit("Set ARCHASTRO_ACCESS_TOKEN before running this example.") + + async with AsyncPlatformClient.with_token( + args.api_key, + args.access_token, + base_url=args.base_url, + ) as client: + socket = await client.open_socket(url=args.ws_url) + session: ThreadChatSession | None = None + + try: + if args.team_id: + channel = await ApiChatChannel.join_team_thread( + socket, + args.team_id, + args.thread_id, + include_metadata=True, + limit=args.limit, + ) + else: + channel = await ApiChatChannel.join_user_thread( + socket, + args.thread_id, + include_metadata=True, + limit=args.limit, + ) + + session = ThreadChatSession(channel) + tui = ThreadChatTui(session, thread_id=args.thread_id, team_id=args.team_id) + session.on_message_added(tui.add_message_payload) + + try: + history = await session.load_history() + except Exception as exc: + history = [] + tui.set_status(f"Connected. History unavailable: {exc}") + tui.add_history(history) + + await tui.run() + finally: + if session is not None: + await session.close() + + +def _env(*names: str) -> str | None: + for name in names: + value = os.environ.get(name) + if value: + return value + return None + + +def main() -> None: + asyncio.run(chat(parse_args())) + + +if __name__ == "__main__": + main() diff --git a/tests/examples/test_create_agent_cli.py b/tests/examples/test_create_agent_cli.py new file mode 100644 index 0000000..840d95d --- /dev/null +++ b/tests/examples/test_create_agent_cli.py @@ -0,0 +1,116 @@ +from __future__ import annotations + +import importlib.util +from pathlib import Path + + +def load_example_module(): + module_path = Path(__file__).parents[2] / "examples" / "create_agent_cli" / "main.py" + spec = importlib.util.spec_from_file_location("create_agent_cli_main", module_path) + assert spec is not None + assert spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def test_parse_args_builds_minimal_agent_input(): + module = load_example_module() + args = module.parse_args( + [ + "--name", + "Demo Agent", + "--identity", + "You are a concise support agent.", + ] + ) + + assert module.build_agent_input(args) == { + "name": "Demo Agent", + "identity": "You are a concise support agent.", + } + + +def test_build_agent_input_includes_optional_scope_and_metadata(): + module = load_example_module() + args = module.parse_args( + [ + "--name", + "Demo Agent", + "--identity", + "You help with onboarding.", + "--model", + "openai/gpt-4.1-mini", + "--org", + "org_123", + "--team", + "team_123", + "--lookup-key", + "demo-agent", + "--metadata-json", + '{"source":"python-sdk-example"}', + ] + ) + + assert module.build_agent_input(args) == { + "name": "Demo Agent", + "identity": "You help with onboarding.", + "model": "openai/gpt-4.1-mini", + "org": "org_123", + "team": "team_123", + "lookup_key": "demo-agent", + "metadata": {"source": "python-sdk-example"}, + } + + +def test_create_agent_uses_env_token_base_url_and_built_payload(monkeypatch): + module = load_example_module() + created_payloads = [] + clients = [] + + class FakeAgents: + def create(self, payload): + created_payloads.append(payload) + return {"id": "agt_123", "name": payload["name"]} + + class FakePlatformClient: + def __init__(self, *, api_key, access_token, base_url): + self.api_key = api_key + self.base_url = base_url + self.access_token = access_token + self.agents = FakeAgents() + self.closed = False + clients.append(self) + + @classmethod + def with_token(cls, api_key, access_token, *, base_url=None): + return cls(api_key=api_key, access_token=access_token, base_url=base_url) + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + self.closed = True + + monkeypatch.setenv("ARCHASTRO_ACCESS_TOKEN", "sat_test") + monkeypatch.setenv("ARCHASTRO_API_KEY", "pk_test") + monkeypatch.setattr(module, "PlatformClient", FakePlatformClient) + args = module.parse_args( + [ + "--name", + "Demo Agent", + "--identity", + "You help users.", + "--base-url", + "http://localhost:4000", + ] + ) + + result = module.create_agent(args) + + assert result == {"id": "agt_123", "name": "Demo Agent"} + assert clients[0].api_key == "pk_test" + assert clients[0].base_url == "http://localhost:4000" + assert clients[0].access_token == "sat_test" + assert clients[0].closed is True + assert created_payloads == [{"name": "Demo Agent", "identity": "You help users."}] diff --git a/tests/examples/test_org_system_user_token.py b/tests/examples/test_org_system_user_token.py new file mode 100644 index 0000000..688cdaf --- /dev/null +++ b/tests/examples/test_org_system_user_token.py @@ -0,0 +1,54 @@ +from __future__ import annotations + +import importlib.util +from pathlib import Path + + +def load_example_module(): + module_path = Path(__file__).parents[2] / "examples" / "org_system_user_token" / "main.py" + spec = importlib.util.spec_from_file_location("org_system_user_token_main", module_path) + assert spec is not None + assert spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def test_main_uses_access_token_env_and_prints_authenticated_user(monkeypatch, capsys): + module = load_example_module() + clients = [] + + class FakeUsers: + def me(self): + return { + "id": "usr_system", + "is_system_user": True, + "org_role": "member", + } + + class FakePlatformClient: + def __init__(self, *, base_url, access_token): + self.base_url = base_url + self.access_token = access_token + self.users = FakeUsers() + clients.append(self) + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + pass + + monkeypatch.setenv("ARCHASTRO_ACCESS_TOKEN", "sat_test") + monkeypatch.setenv("ARCHASTRO_PLATFORM_BASE_URL", "http://localhost:4000") + monkeypatch.setattr(module, "PlatformClient", FakePlatformClient) + + module.main() + + assert clients[0].base_url == "http://localhost:4000" + assert clients[0].access_token == "sat_test" + assert capsys.readouterr().out.splitlines() == [ + "Authenticated as usr_system", + "System user: True", + "Org role: member", + ] diff --git a/tests/examples/test_thread_chat_tui.py b/tests/examples/test_thread_chat_tui.py new file mode 100644 index 0000000..b4bbeb1 --- /dev/null +++ b/tests/examples/test_thread_chat_tui.py @@ -0,0 +1,297 @@ +from __future__ import annotations + +import importlib.util +import sys +from pathlib import Path + +import pytest + + +def load_example_module(): + module_path = Path(__file__).parents[2] / "examples" / "thread_chat_tui" / "main.py" + spec = importlib.util.spec_from_file_location("thread_chat_tui_main", module_path) + assert spec is not None + assert spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +def test_parse_args_uses_platform_env_names(monkeypatch): + module = load_example_module() + + monkeypatch.setenv("ARCHASTRO_PLATFORM_BASE_URL", "http://localhost:4005") + monkeypatch.setenv("ARCHASTRO_PLATFORM_WS_URL", "ws://localhost:4005/socket/api/websocket") + + args = module.parse_args(["thr_123"]) + + assert args.base_url == "http://localhost:4005" + assert args.ws_url == "ws://localhost:4005/socket/api/websocket" + + +def test_parse_args_accepts_positional_thread_and_team_scope(): + module = load_example_module() + + args = module.parse_args(["thr_123", "--team", "team_123", "--limit", "10"]) + + assert args.thread_id == "thr_123" + assert args.team_id == "team_123" + assert args.limit == 10 + + +def test_message_from_payload_prefers_actor_name_and_content(): + module = load_example_module() + + message = module.message_from_payload( + { + "message": { + "id": "msg_123", + "actors": [{"name": "Ada Lovelace", "id": "user_123"}], + "content": "hello from the websocket", + "created_at": "2026-06-11T12:00:00Z", + } + } + ) + + assert message.id == "msg_123" + assert message.author == "Ada Lovelace" + assert message.content == "hello from the websocket" + + +def test_render_message_wraps_for_terminal_width(): + module = load_example_module() + message = module.ChatMessage( + id="msg_123", + author="Ada", + content="one two three four five six", + created_at=None, + ) + + lines = module.render_message(message, width=18) + + assert lines == ["Ada: one two three", " four five six"] + + +def test_draw_avoids_bottom_right_curses_cell(): + module = load_example_module() + tui = module.ThreadChatTui(FakeSession(), thread_id="thr_123", team_id=None) + tui._draft = "hello from the bottom row" + + screen = FakeCursesScreen(module.curses, height=8, width=32) + + tui._draw(screen) + + bottom_writes = [write for write in screen.writes if write[0] == 7] + assert bottom_writes + assert all(write[3] < 32 for write in bottom_writes) + + +@pytest.mark.asyncio +async def test_tui_sends_messages_through_session_boundary(): + module = load_example_module() + session = FakeSession() + tui = module.ThreadChatTui(session, thread_id="thr_123", team_id=None) + + await tui._post_message("hello through the session") + + assert session.sent == [ + { + "content": "hello through the session", + "idempotency_key": tui._messages[0].idempotency_key, + } + ] + assert tui._messages[0].state == "sent" + assert tui._status == "Message sent." + + +@pytest.mark.asyncio +async def test_chat_joins_user_thread_and_cleans_up(monkeypatch): + module = load_example_module() + + socket = FakeSocket() + channel = FakeChannel() + clients = [] + joins = [] + + class FakeAsyncPlatformClient: + def __init__(self, *, api_key, access_token, base_url): + self.api_key = api_key + self.access_token = access_token + self.base_url = base_url + self.open_socket_calls = [] + self.closed = False + clients.append(self) + + @classmethod + def with_token(cls, api_key, access_token, *, base_url=None): + return cls(api_key=api_key, access_token=access_token, base_url=base_url) + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + self.closed = True + + async def open_socket(self, *, url=None): + self.open_socket_calls.append({"url": url}) + return socket + + class FakeApiChatChannel: + @staticmethod + async def join_user_thread(socket_arg, thread_id, *, include_metadata, limit): + joins.append((socket_arg, thread_id, include_metadata, limit)) + return channel + + monkeypatch.setenv("ARCHASTRO_ACCESS_TOKEN", "sat_test") + monkeypatch.setenv("ARCHASTRO_API_KEY", "pk_test") + monkeypatch.setattr(module, "AsyncPlatformClient", FakeAsyncPlatformClient) + monkeypatch.setattr(module, "ApiChatChannel", FakeApiChatChannel) + monkeypatch.setattr(module, "ThreadChatTui", FakeTui) + + args = module.parse_args( + [ + "thr_123", + "--base-url", + "http://localhost:4000", + "--ws-url", + "ws://localhost:4000/socket/api/websocket", + "--limit", + "5", + ] + ) + + await module.chat(args) + + assert clients[0].api_key == "pk_test" + assert clients[0].access_token == "sat_test" + assert clients[0].base_url == "http://localhost:4000" + assert clients[0].open_socket_calls == [{"url": "ws://localhost:4000/socket/api/websocket"}] + assert clients[0].closed is True + assert joins == [(socket, "thr_123", True, 5)] + assert isinstance(FakeTui.instances[0].session, module.ThreadChatSession) + assert channel.message_handler is not None + assert channel.left is True + + +@pytest.mark.asyncio +async def test_chat_closes_client_when_leave_raises(monkeypatch): + module = load_example_module() + + socket = FakeSocket() + channel = FakeChannel(leave_error=RuntimeError("leave failed")) + clients = [] + + class FakeAsyncPlatformClient: + @classmethod + def with_token(cls, api_key, access_token, *, base_url=None): + client = cls() + clients.append(client) + return client + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + self.closed = True + + async def open_socket(self, *, url=None): + return socket + + class FakeApiChatChannel: + @staticmethod + async def join_user_thread(socket_arg, thread_id, *, include_metadata, limit): + return channel + + monkeypatch.setenv("ARCHASTRO_ACCESS_TOKEN", "sat_test") + monkeypatch.setenv("ARCHASTRO_API_KEY", "pk_test") + monkeypatch.setattr(module, "AsyncPlatformClient", FakeAsyncPlatformClient) + monkeypatch.setattr(module, "ApiChatChannel", FakeApiChatChannel) + monkeypatch.setattr(module, "ThreadChatTui", FakeTui) + + args = module.parse_args(["thr_123"]) + + with pytest.raises(RuntimeError, match="leave failed"): + await module.chat(args) + + assert clients[0].closed is True + + +class FakeSocket: + pass + + +class FakeChannel: + def __init__(self, *, leave_error=None): + self.leave_error = leave_error + self.message_handler = None + self.left = False + + def on_message_added(self, callback): + self.message_handler = callback + + async def api_chat_list_messages(self, payload): + return {"response": {"messages": []}} + + async def leave(self): + self.left = True + if self.leave_error: + raise self.leave_error + + +class FakeSession: + def __init__(self): + self.sent = [] + + async def send_message(self, content, *, idempotency_key): + self.sent.append({"content": content, "idempotency_key": idempotency_key}) + + +class FakeTui: + instances = [] + + def __init__(self, session, *, thread_id, team_id): + self.session = session + self.thread_id = thread_id + self.team_id = team_id + self.history = None + FakeTui.instances.append(self) + + def add_history(self, messages): + self.history = messages + + def add_message_payload(self, payload): + pass + + def set_status(self, status): + pass + + async def run(self): + pass + + +class FakeCursesScreen: + def __init__(self, curses_module, *, height, width): + self._curses = curses_module + self._height = height + self._width = width + self.writes = [] + self.cursor = None + self.refreshed = False + + def getmaxyx(self): + return self._height, self._width + + def erase(self): + pass + + def addnstr(self, y, x, text, n, *attrs): + if y == self._height - 1 and x + n >= self._width: + raise self._curses.error("addnwstr() returned ERR") + self.writes.append((y, x, text, n, attrs)) + + def move(self, y, x): + self.cursor = (y, x) + + def refresh(self): + self.refreshed = True From c7817924124e306db2513f343bd388c87d816c42 Mon Sep 17 00:00:00 2001 From: Calvin Grunewald Date: Fri, 12 Jun 2026 14:21:01 -0700 Subject: [PATCH 2/3] update archagents readme quickstart --- README.md | 139 +++++++++++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 133 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index a0ac45b..aece40e 100644 --- a/README.md +++ b/README.md @@ -1,18 +1,145 @@ -# archastro-python +# ArchAstro Python SDK -Python SDK for the ArchAstro Platform API. +Python SDK for the ArchAstro Platform API and ArchAgents runtime APIs. ```bash -uv add archastro-sdk # or: pip install archastro-sdk +uv add archastro-sdk +# or +pip install archastro-sdk ``` +The clients default to the production API gateway, `https://platform.archastro.ai`. +Set `ARCHASTRO_PLATFORM_BASE_URL` only when targeting local development, +staging, or another non-production environment. + +## Getting Started + +Choose the auth path that matches how your Python process should run. + +### ArchAgents Org Bot or Worker + +Use this path for ArchAgents bots, background workers, cron jobs, ingestion +jobs, and integrations that should act as an org-owned system user. Your Python +process only needs a system-user access token: + +```bash +export ARCHASTRO_ACCESS_TOKEN=sat_... +``` + +Create that token with `archagent` while logged in as an org admin: + +```bash +archagent auth login +export ARCHASTRO_ORG_ID=org_... + +export ARCHASTRO_SYSTEM_USER_ID="$( + archagent --json create user \ + --system-user \ + --name "Python SDK Bot" \ + --org "$ARCHASTRO_ORG_ID" \ + --org-role member | + jq -r '.id' +)" + +export ARCHASTRO_ACCESS_TOKEN="$( + archagent --json create usertoken \ + --user "$ARCHASTRO_SYSTEM_USER_ID" \ + --name "python-sdk-service" | + jq -r '.token' +)" +``` + +Use the sync client for scripts and CLIs: + +```python +import os + +from archastro.platform import PlatformClient + +with PlatformClient(access_token=os.environ["ARCHASTRO_ACCESS_TOKEN"]) as client: + user = client.users.me() + +print(user["id"], user.get("is_system_user")) +``` + +Use the async client inside async services or workers: + +```python +import asyncio +import os + +from archastro.platform import AsyncPlatformClient + + +async def main() -> None: + async with AsyncPlatformClient( + access_token=os.environ["ARCHASTRO_ACCESS_TOKEN"], + ) as client: + user = await client.users.me() + + print(user["id"], user.get("is_system_user")) + + +asyncio.run(main()) +``` + +See [`examples/org_system_user_token`](examples/org_system_user_token) for the +complete system-user walkthrough. + +### Developer App Auth + +Use this path when you already have a publishable API key and a user access +token from a developer app login flow. + +```bash +export ARCHASTRO_API_KEY=pk_... +export ARCHASTRO_ACCESS_TOKEN=sat_... +``` + +```python +import os + +from archastro.platform import PlatformClient + +client = PlatformClient.with_token( + os.environ["ARCHASTRO_API_KEY"], + os.environ["ARCHASTRO_ACCESS_TOKEN"], +) + +with client: + teams = client.teams.list() +``` + +Async setup uses the same factory: + ```python -from archastro import ArchAstro +import asyncio +import os -client = ArchAstro(api_key="pk_...") -teams = client.v1.teams.list() +from archastro.platform import AsyncPlatformClient + + +async def main() -> None: + async with AsyncPlatformClient.with_token( + os.environ["ARCHASTRO_API_KEY"], + os.environ["ARCHASTRO_ACCESS_TOKEN"], + ) as client: + teams = await client.teams.list() + print(teams) + + +asyncio.run(main()) ``` +## Examples + +- [`examples/org_system_user_token`](examples/org_system_user_token) — run the + SDK as an ArchAgents org-owned system user. +- [`examples/create_agent_cli`](examples/create_agent_cli) — wrap the sync SDK + in a small CLI that creates an agent. +- [`examples/thread_chat_tui`](examples/thread_chat_tui) — chat in an existing + thread from a terminal UI using the async websocket helpers. + ## Packages All public code lives under the single top-level `archastro` package: From 7ccff51e03b1a7e739b86f513691851a03c12a28 Mon Sep 17 00:00:00 2001 From: Calvin Grunewald Date: Fri, 12 Jun 2026 14:29:54 -0700 Subject: [PATCH 3/3] clarify ArchAgents README setup --- README.md | 12 ++++++--- examples/org_system_user_token/README.md | 34 ++++++++++++++---------- 2 files changed, 29 insertions(+), 17 deletions(-) diff --git a/README.md b/README.md index aece40e..a9b5ab4 100644 --- a/README.md +++ b/README.md @@ -26,11 +26,17 @@ process only needs a system-user access token: export ARCHASTRO_ACCESS_TOKEN=sat_... ``` -Create that token with `archagent` while logged in as an org admin: +Create that token with `archagent` while logged in as an org admin. Replace +`user@company.com` with your ArchAgents login email. The setup is grouped as +one shell block so GitHub's copy button copies the full sequence: ```bash -archagent auth login -export ARCHASTRO_ORG_ID=org_... +archagent auth login user@company.com + +export ARCHASTRO_ORG_ID="$( + archagent describe me --json | + jq -er '.session.org' +)" export ARCHASTRO_SYSTEM_USER_ID="$( archagent --json create user \ diff --git a/examples/org_system_user_token/README.md b/examples/org_system_user_token/README.md index 59d8ad5..b0a9d0e 100644 --- a/examples/org_system_user_token/README.md +++ b/examples/org_system_user_token/README.md @@ -77,17 +77,18 @@ asyncio.run(main()) ## If You Are an ArchAgents User If you are already using ArchAgents and can log in as an org admin, create the -system user and token with `archagent`. +system user and token with `archagent`. Replace `user@company.com` with your +ArchAgents login email. GitHub adds a copy button to fenced code blocks, so this +setup is kept as one complete shell block. ```bash -archagent auth login -export ARCHASTRO_ORG_ID=org_... -``` +archagent auth login user@company.com -Create an org-scoped system user. `member` is the safest default role for most -bots and integrations. +export ARCHASTRO_ORG_ID="$( + archagent describe me --json | + jq -er '.session.org' +)" -```bash export ARCHASTRO_SYSTEM_USER_ID="$( archagent --json create user \ --system-user \ @@ -96,12 +97,7 @@ export ARCHASTRO_SYSTEM_USER_ID="$( --org-role member | jq -r '.id' )" -``` - -Create a token for that system user. The raw token is shown once, so put it in -your secret manager immediately. -```bash export ARCHASTRO_ACCESS_TOKEN="$( archagent --json create usertoken \ --user "$ARCHASTRO_SYSTEM_USER_ID" \ @@ -110,6 +106,10 @@ export ARCHASTRO_ACCESS_TOKEN="$( )" ``` +The system user is org-scoped. `member` is the safest default role for most bots +and integrations. The raw token is shown once, so put +`ARCHASTRO_ACCESS_TOKEN` in your secret manager immediately. + Now run your Python service with `ARCHASTRO_ACCESS_TOKEN` in its environment. Only set `ARCHASTRO_PLATFORM_BASE_URL` when targeting local development, staging, or another non-production environment. @@ -118,11 +118,17 @@ staging, or another non-production environment. Use `archastro` when you need to bootstrap the system user from a developer app context instead of an ArchAgents org session. Run from an initialized ArchAstro -project, or pass `--app ` to each command. +project, or pass `--app ` to each command. Replace `Example Org` with +the org name, slug, or domain you want to bootstrap. ```bash archastro auth login -export ARCHASTRO_ORG_ID=org_... + +export ARCHASTRO_ORG_SEARCH="Example Org" +export ARCHASTRO_ORG_ID="$( + archastro --json list orgs --search "$ARCHASTRO_ORG_SEARCH" | + jq -er '.data[0].id' +)" export ARCHASTRO_SYSTEM_USER_ID="$( archastro --json create user \