Build a Connector

tools/aos/aos-{name}/
├── connector.json                    <- Authoritative metadata
├── README.md                         <- Operator docs
└── agent-harness/
    ├── permissions.json              <- Mode-gate map
    ├── pyproject.toml                <- Python config + CLI entry point
    ├── cli_aos/
    │   └── {name}/
    │       ├── __init__.py           <- Version marker
    │       ├── constants.py          <- Env var names, API base URLs
    │       ├── errors.py             <- CliError dataclass
    │       ├── output.py             <- JSON/text helpers
    │       ├── config.py             <- Auth resolution from env vars
    │       ├── client.py             <- HTTP client (actual API calls)
    │       ├── runtime.py            <- Command functions + health
    │       └── cli.py                <- Click CLI wiring
    └── tests/
        ├── conftest.py               <- sys.path setup
        └── test_cli.py               <- FakeClient mocking

cp -r tools/aos/templates/python-click-tool tools/aos/aos-{name}
cd tools/aos/aos-{name}

cp -r tools/aos/aos-twilio tools/aos/aos-{name}

{
  "tool": "aos-{name}",
  "backend": "{vendor}-api",
  "manifest_schema_version": "1.0.0",
  "connector": {
    "label": "My Vendor",
    "category": "category-name",
    "categories": ["primary-category", "secondary"],
    "resources": ["resource1", "resource2"]
  },
  "auth": {
    "kind": "service-key",
    "required": true,
    "service_keys": ["VENDOR_API_KEY"],
    "interactive_setup": [
      "Get an API key from vendor.com/settings",
      "Add VENDOR_API_KEY in API Keys settings"
    ]
  },
  "commands": [
    {
      "id": "items.list",
      "summary": "List items",
      "required_mode": "readonly",
      "supports_json": true,
      "resource": "items",
      "action_class": "read"
    },
    {
      "id": "items.create",
      "summary": "Create a new item",
      "required_mode": "write",
      "supports_json": true,
      "resource": "items",
      "action_class": "write"
    }
  ]
}

{
  "commands": {
    "health": "readonly",
    "config.show": "readonly",
    "capabilities": "readonly",
    "items.list": "readonly",
    "items.get": "readonly",
    "items.create": "write",
    "items.update": "write",
    "items.delete": "admin"
  }
}

import json
import urllib.request
import urllib.error

class Client:
    def __init__(self, api_key: str, base_url: str = "https://api.vendor.com/v1"):
        self.api_key = api_key
        self.base_url = base_url

    def _request(self, method: str, path: str, data=None):
        url = f"{self.base_url}{path}"
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        }
        body = json.dumps(data).encode() if data else None
        req = urllib.request.Request(url, data=body, headers=headers, method=method)
        try:
            with urllib.request.urlopen(req) as resp:
                return json.loads(resp.read().decode())
        except urllib.error.HTTPError as e:
            raise RuntimeError(f"API error {e.code}: {e.read().decode()}")

    def list_items(self, limit=30):
        return self._request("GET", f"/items?limit={limit}")

    def create_item(self, title: str, description: str = ""):
        return self._request("POST", "/items", {"title": title, "description": description})

from .client import Client
from .config import resolve_config
from .output import success, error

def health():
    cfg = resolve_config()
    if not cfg.get("api_key"):
        return {"status": "needs_setup", "checks": {"auth": False}, "next_steps": ["Set VENDOR_API_KEY"]}
    try:
        client = Client(cfg["api_key"])
        client.list_items(limit=1)
        return {"status": "healthy", "checks": {"auth": True, "api": True}}
    except Exception as e:
        return {"status": "degraded", "checks": {"auth": True, "api": False}, "detail": str(e)}

def items_list(mode, limit=30):
    cfg = resolve_config()
    client = Client(cfg["api_key"])
    return success("items.list", client.list_items(limit))

import click
from .runtime import health, items_list
from .output import format_output

@click.group()
@click.option("--json", "use_json", is_flag=True)
@click.option("--mode", default="readonly")
@click.option("--verbose", is_flag=True)
@click.pass_context
def cli(ctx, use_json, mode, verbose):
    ctx.ensure_object(dict)
    ctx.obj["json"] = use_json
    ctx.obj["mode"] = mode

@cli.command()
@click.pass_context
def health_cmd(ctx):
    """Report connector health."""
    click.echo(format_output(health(), ctx.obj["json"]))

# ... more commands

[project.scripts]
aos-{name} = "cli_aos.{name}.cli:cli"

cd agent-harness
python3 -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"
python -m pytest tests/ -v

# Check capabilities
aos-{name} --json capabilities | python -m json.tool

# Check health
aos-{name} --json health | python -m json.tool

# Run tests
python -m pytest tests/ -v