> ## Documentation Index
> Fetch the complete documentation index at: https://docs.argentos.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Build a Connector

> Step-by-step guide to creating a new AOS connector from scratch.

## Prerequisites

* Python 3.11+
* Target vendor API docs reviewed
* ArgentOS repo cloned (`tools/aos/` directory available)

## Directory Structure

Every connector lives at `tools/aos/aos-{name}/`:

```
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
```

## Step 1: Copy the Reference Template

```bash theme={null}
cp -r tools/aos/templates/python-click-tool tools/aos/aos-{name}
cd tools/aos/aos-{name}
```

Or copy from an existing connector as a reference:

```bash theme={null}
cp -r tools/aos/aos-twilio tools/aos/aos-{name}
```

Rename all internal references from the source connector to your new name.

## Step 2: Write `connector.json`

This is the authoritative manifest. It tells ArgentOS what your connector does, what commands it supports, what auth it needs, and how to render it in the dashboard.

```json theme={null}
{
  "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"
    }
  ]
}
```

## Step 3: Write `permissions.json`

Map every CLI command to a permission tier:

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

## Step 4: Implement `client.py`

This is the core — the HTTP client that calls the actual vendor API.

```python theme={null}
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})
```

Guidelines:

* Use stdlib `urllib` (no external HTTP dependencies)
* One method per command
* Return structured dicts, not raw responses
* Auth resolved from constructor params (injected by `config.py`)

## Step 5: Implement `runtime.py`

Command result functions and the three required contracts:

```python theme={null}
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))
```

## Step 6: Implement `cli.py`

Wire commands to runtime functions using Click:

```python theme={null}
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
```

Entry point in `pyproject.toml`:

```toml theme={null}
[project.scripts]
aos-{name} = "cli_aos.{name}.cli:cli"
```

## Step 7: Write Tests

```bash theme={null}
cd agent-harness
python3 -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"
python -m pytest tests/ -v
```

Required test coverage:

* Commands match `connector.json` manifest
* Health: missing creds returns `needs_setup`, valid creds returns `healthy`
* Read commands return expected data shapes (use FakeClient)
* Write commands respect mode gating (calling write in readonly mode fails)

## Step 8: Verify End-to-End

```bash theme={null}
# 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
```

## Certification Gates

Before a connector is production-ready:

| Gate         | Requirement                                                             |
| ------------ | ----------------------------------------------------------------------- |
| **Contract** | Valid `connector.json`, working `capabilities`, `health`, `config show` |
| **Safety**   | Write actions classified, destructive actions gated at `admin` mode     |
| **Audit**    | Tool calls preserve evidence, failures return understandable errors     |
| **Docs**     | README exists with setup instructions and known limitations             |
| **Tests**    | All required test categories pass                                       |

## Risk Tiers

| Tier                 | Access Level            | Examples                           | Review Required   |
| -------------------- | ----------------------- | ---------------------------------- | ----------------- |
| 1: Read-heavy        | Queries only            | Search tickets, list contacts      | Minimal           |
| 2: Bounded write     | Scoped mutations        | Reply to message, create event     | Standard          |
| 3: Business mutation | Business state changes  | CRM stage change, invoice creation | Thorough          |
| 4: Destructive/admin | Irreversible operations | Delete records, revoke access      | Explicit approval |