Metadata-Version: 2.5
Name: signalcore
Version: 4.0.0
Summary: SignalCore's server SDK: agents, conversations and everything of the public API, from your backend.
Project-URL: Homepage, https://signalcore.ai
Project-URL: Documentation, https://signalcore.ai/docs
Project-URL: Repository, https://github.com/OmnistGroup/signalcore
Author-email: SignalCore <ismael@signalcore.ai>
License-Expression: MIT
License-File: LICENSE
Keywords: agents,ai,sdk,signalcore
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: httpx>=0.27
Requires-Dist: typing-extensions>=4.7
Requires-Dist: websockets>=13
Description-Content-Type: text/markdown

# signalcore

SignalCore's server SDK for Python: your backend makes organizations for your customers, sets up
their agents, talks to them and hears what happens, through SignalCore's public API
(`https://signalcore.ai/v1`). Every operation of the API is a typed method; the SDK holds no logic
of its own.

```bash
pip install "signalcore>=3,<4"     # uv add, poetry add alike
```

Python 3.10 or later. Version 4 is for SignalCore 2.0's API as it is now (no squads: the
organization is the agents' one group); 2.x was the same API with squads, 0.x the API it replaced.
Full docs, in Spanish: [signalcore.ai/docs](https://signalcore.ai/docs).

## Keys

A secret key (`sk_…`) is made by the account's owner in the app, in **Tu plataforma → Claves**
(the app shows Tu plataforma to an account of kind "Lo integro en mi producto"; any account can
switch in **Ajustes → General → Tipo de cuenta**). It is shown once: keep it in your server's
environment, never in a page. Each key has scopes:

| Scope | Lets the key |
|---|---|
| `read` | read everything (every `GET`) |
| `write` | change the organization: agents, their folder, files, tables, the vault, automations, triggers, channels, apps, contacts, who answers a conversation, the inbox, your tools |
| `talk` | start conversations, speak in them, call an agent, upload a conversation's files, write as the business on a channel, mint person tokens |
| `admin` | make and delete your customers' organizations, rename your own, deploy and update templates, make and remove webhooks |

A key without the scope gets `403 forbidden`. Revoking a key ends every token it minted.

```python
from signalcore import SignalCore

client = SignalCore()  # reads SIGNALCORE_API_KEY; or SignalCore("sk_…")
```

Keyword options: `org` (act on a customer's organization, below), `base_url`
(`SIGNALCORE_BASE_URL`, or `https://signalcore.ai`), `timeout` in seconds (90), `http_client` (an
`httpx.Client` of your own). `AsyncSignalCore` has every method of `SignalCore`, awaited, and its
`http_client` is an `httpx.AsyncClient`. Both close with `close()` or as a context manager.

## Quickstart

```python
agent = client.agents.create(
    name="Recepción",
    instructions="You answer the clinic's patients briefly, in Spanish.",
)
conversation = client.conversations.start(agent["id"])

reply = client.conversations.say(conversation["id"], text="¿Abrís el sábado?", wait=True)
print(reply["status"], reply["replies"][0]["text"])
```

Answers are plain dicts typed with `TypedDict`s (`signalcore.types`), so your editor knows every
field.

## Agents

```python
agents = client.agents.list()  # the builder first (is_builder), then by name
one = client.agents.get("agt_…")

client.agents.update(
    one["id"],
    instructions="Answer in Spanish, in two sentences at most.",
    model="claude-opus-5-5",  # client.catalog.models() lists the ids
    timezone="Europe/Madrid",
)
client.agents.delete(one["id"])  # the builder can't be deleted
```

`agents.create` takes `name` and, optionally, `role`, `description`, `instructions`, `model` and
`icon`; a name the organization has already is `409 conflict`. `agents.update` changes only what
it is passed, and also takes `voice` (how it talks on calls), `packages` (what its own code
installs), `actions` and `needs_approval`, and what the agent may set up itself in a
conversation: `connects`, the apps it connects with `connect_app` (which `kinds` or `toolkits`,
`whose` account, what it `gets` of one, who gives the yes for its `writes`), and `switches`, the
automations it turns on or off with `switch_automation`:

```python
client.agents.update(
    one["id"],
    actions=[*one["actions"], "connect_app", "switch_automation"],
    connects={
        "kinds": ["crm"],
        "whose": "person",
        "gets": "all",
        "writes": "person",
        "return_url": "https://your.app/chat",  # where the person lands once they authorized it
    },
    switches=[{"agent": "Coach", "automation": "weekly_report"}],  # no agent: its own
)
```

### Knowledge

Files every agent of the organization reads, or one agent's own. `searchable=True` indexes it
(PDF, Word, Excel, PowerPoint, text) so the agent searches it when asked; up to 50 MB a file.

```python
from pathlib import Path

client.files.put(
    "tarifas.pdf",
    Path("tarifas.pdf").read_bytes(),
    searchable=True,
    content_type="application/pdf",
)
client.agents.files_put(
    agent["id"],
    "guion.md",
    "# Cómo saludar\n…".encode(),
    searchable=True,
    content_type="text/markdown",
)

hits = client.files.search(query="precio de una limpieza")  # [{path, location, text, …}]
data = client.files.get("tarifas.pdf")  # bytes
client.files.delete("tarifas.pdf")

# Something the agent keeps in mind, as it would learn it itself
client.agents.memories_create(agent["id"], text="Los sábados abrimos de 10:00 a 14:00.")
```

Tables (`client.data`), keys the agent uses on the web (`client.vault`,
`client.agents.vault_put`) and the apps it uses (`client.connections`) are in the same client.

### The agent's own actions

An agent is a folder: `agent.yaml`, `instructions.md` and its code. A file in `actions/` with one
`@sc.action` function is a tool the model calls while it talks:

```python
PRECIO = '''import sc

PRECIOS = {"limpieza": 45.0, "blanqueamiento": 180.0}


@sc.action
def precio(servicio: str) -> float:
    """El precio de un servicio en euros. Llámala cuando pregunten cuánto cuesta."""
    return PRECIOS[servicio.lower()]
'''

client.agents.folder_put(agent["id"], "actions/precio.py", content=PRECIO)

actions = client.agents.actions_list(agent["id"])  # precio, with source "own"
files = client.agents.folder_list(agent["id"])  # every file of the folder
```

A file that isn't valid code is refused (`422`) with the reason. `automations/<name>.py` is what
the agent does on its own; see `client.automations` (`test`, `switch`) and `client.runs`.

## Conversations

```python
conversation = client.conversations.start(agent["id"])
reply = client.conversations.say(conversation["id"], text="Hola", wait=True)

if reply["status"] == "done":
    print([message["text"] for message in reply["replies"]])
elif reply["status"] == "waiting":
    print(reply["waiting"])  # [{id, action, kind: ask | approve | tool, decides}]
elif reply["status"] == "failed":
    print(reply["error"])  # {code, detail}
else:  # "running": still going after a minute; read it later
    ...

whole = client.conversations.get(conversation["id"])  # with its messages
latest = client.conversations.list(agent["id"])  # newest first
```

With `wait=True` the call answers when the agent's turn ends, up to a minute. Without it, it
answers at once with `running`; the reply comes on the live socket or a webhook. While a turn runs,
another `say` is `422 busy`; while the turn waits on a question, `422 pending`.

When an action waits for the business's yes (`decides: "business"`, also told by the
`approval.requested` webhook), your backend gives it:

```python
client.conversations.decide(conversation["id"], request=reply["waiting"][0]["id"], approve=True)
```

### Files in a conversation

Upload first, then name the file in `say` (up to 10 a message):

```python
invoice = client.conversations.files_create(
    conversation["id"],
    Path("factura.pdf").read_bytes(),
    file_name="factura.pdf",
    content_type="application/pdf",
)
client.conversations.say(
    conversation["id"], text="¿Me explicas esta factura?", files=[invoice["id"]], wait=True
)
```

### Live

Every frame of a conversation as it happens:

```python
with client.conversations.live(conversation["id"]) as live:
    live.say("Hola")
    for frame in live:
        if frame["type"] == "text":
            print(frame["delta"], end="", flush=True)
        if frame["type"] == "done":
            break
```

With `AsyncSignalCore`, the same with `async with`, `await live.say(…)` and `async for`.

What comes down: `working` (a turn began), `message` (a message stored, in `message`), `text` (a
piece of the reply, in `delta`), `step` (an action: `running`, then `done`, `failed` or
`waiting`), `ui` (a screen the agent drew), `request` (an action that asks first), `tool` (a
client tool for the person's page), `handoff` (it went to a person), `holder` (who answers
changed), `done` (the turn is over), `stopped`, and `error` (`code`, `detail`). What you send:
`live.say(text, file_ids)`, `live.stop()`, or any frame with `live.send({"type": …})`. Closing
the socket doesn't stop the reply: it is stored.

## Your customers' organizations

An agency or a product makes one organization per customer (`admin` scope). Each comes with its
own builder and shares nothing with the others. Name the agents it starts with: your agents that
are templates (below).

```python
customer = client.orgs.create(name="Clínica Norte", templates=[agent["id"]])
# {id, name, slug: "clinica-norte", kind: "business", agents: [ … ]}

# Act on it: the same key, with `org` as its id or its slug (SignalCore-Org)
theirs = SignalCore(org=customer["slug"])
theirs.agents.list()  # their own agents
theirs.files.put("carta.pdf", data, searchable=True, content_type="application/pdf")

every = client.orgs.list()  # oldest first
same = client.orgs.get(customer["id"])
client.orgs.delete(customer["id"])  # 409 while it has channels or connected apps left
```

What each customer used this month, for your bill (`from_` and `to`, a `datetime` or ISO 8601,
for another stretch):

```python
for used in client.usage.managed():
    print(used["org_id"], used["cost"])  # USD
```

## Templates

One of your agents, given to your customers: each customer's organization gets its own copy, and
your changes reach every copy except what that customer changed.

```python
client.templates.deploy(agent["id"])  # version 1, as the agent is now

# Your deploy owns these fields: from now on only it writes them, in every copy
client.templates.deploy(
    agent["id"],
    spec={"instructions": "…", "files": {"actions/precio.py": PRECIO}},
)

client.agents.update(agent["id"], model="claude-opus-5-5")
status = client.templates.update_all(agent["id"])  # a new version, brought to every copy
# status["agents"]: [{org_id, org_name, agent_id, version, overrides}]

theirs.templates.make(agent["id"])  # a copy in an organization that exists (SignalCore-Org)
client.templates.get(agent["id"])  # the template and every copy
client.templates.delete(agent["id"])  # no longer a template; the copies stay
```

`spec` takes any of the agent's fields: `name`, `role`, `description`, `icon`, `instructions`,
`model`, `timezone`, `voice`, `packages`, `actions`, `asks`, `access`, `tools`, `files`.

## People in your product

Your backend mints a token for each user (an hour, the agents it names; `talk` scope), and your
page talks with it through [`@signalcore/ui`](https://www.npmjs.com/package/@signalcore/ui):

```python
token = client.tokens.create(
    agents=[agent["id"]], external_id=user.id, name=user.name, email=user.email
)
return {"token": token["token"], "expires_at": token["expires_at"]}
```

For anonymous visitors of a site, connect the agent's web channel; its publishable key (`pk_…`)
goes in the page, and works only from the origins named:

```python
connected = client.channels.web_connect(agent_id=agent["id"], origins=["https://clinicanorte.com"])
key = connected["channel"]["web"]["key"]
# <script src="https://signalcore.ai/v1/widget.js" data-key="pk_…" async></script>
```

## Contacts

The people your agents serve: who writes or calls on a channel, or a user of your product with a
token. A conversation names its contact:

```python
conversation = client.conversations.get("cnv_…")
if conversation.get("contact"):
    contact_id = conversation["contact"]["id"]
    contact = client.contacts.get(contact_id)  # {id, name, muted, opted_out, created_at}
    memories = client.contacts.memories(contact_id)  # what the agents keep about them
    tasks = client.contacts.tasks(contact_id)  # what they asked for later
    their_conversations = client.contacts.conversations(contact_id)
    client.contacts.update(contact_id, muted=True)  # the agent stops answering them
```

What is one person's is kept on their contact, and an agent uses it only where it works for
them, never with anyone else: their files (what they sent on any channel lands in `received/`),
their keys of the vault (names only come back) and what the agents know about them.

```python
client.contacts.files_put(contact_id, "poliza.pdf", data, searchable=True, content_type="application/pdf")
client.contacts.files(contact_id)  # [{path, media_type, size, searchable, …}]
client.contacts.files_get(contact_id, "received/foto.jpg")  # bytes
client.contacts.vault_put(contact_id, "portal", value="…", hosts=["portal.example.com"])
client.contacts.memories_create(contact_id, text="Prefiere las tardes.", agent_id=agent["id"])
```

## Your tools

Functions of your own the agent may call. A **server** tool is a URL of yours that SignalCore
POSTs to, signed; a **client** tool runs in the person's page (`@signalcore/ui` answers it).

```python
declared = client.tools.put(
    agent["id"],
    "order_status",
    description="The status of an order. Call it when the person asks where their order is.",
    input_schema={
        "type": "object",
        "properties": {"order_id": {"type": "string"}},
        "required": ["order_id"],
    },
    mode="server",
    url="https://api.tuapp.com/signalcore/order-status",
)
# declared["secret"]: the organization's tool secret, only with its first server tool
# (a new one: client.tools.rotate_secret())
```

Each call is `{tool, input, conversation_id, org_id, person}`, signed with the tool secret as a
webhook is, so the same `verify` checks it. Answer JSON within 30 seconds:

```python
from signalcore import SignatureError, verify


@app.post("/signalcore/order-status")
async def order_status(request: Request):
    try:
        call = verify(await request.body(), request.headers, TOOL_SECRET)
    except SignatureError:
        raise HTTPException(400)
    return await orders.status(call["input"]["order_id"])
```

Your API has an OpenAPI document? Import it and pick operations instead:
`client.tools.import_openapi(agent["id"], url=…)`, then
`client.tools.pick_openapi(agent["id"], picked=["listOrders"])`.

## Webhooks

```python
hook = client.webhooks.create(
    url="https://api.tuapp.com/signalcore",
    events=["message.created", "approval.requested", "run.failed"],
)
# hook["secret"] (whsec_…): shown this once
```

Events: `message.created`, `conversation.holder_changed`, `inbox.item_created`, `run.finished`,
`run.failed`, `call.ended`, `channel.status_changed`, `approval.requested`. Each arrives as
`{id, type, created_at, org_id, data}`; check it before trusting it, with the body exactly as it
came:

```python
from signalcore import SignatureError, verify


@app.post("/signalcore")
async def hook(request: Request):
    try:
        event = verify(await request.body(), request.headers, WEBHOOK_SECRET)
    except SignatureError:
        raise HTTPException(400)
    if event["type"] == "message.created":
        ...
    return Response(status_code=204)
```

Answer a 2xx within 10 seconds. Otherwise the delivery is tried again (after 15 s, 30, 60 and
120) and is `dead` after the fifth try; `id` is the same each try. To send one again:

```python
for delivery in client.webhooks.deliveries(hook["id"]):
    if delivery["status"] == "dead":
        client.webhooks.retry(hook["id"], delivery["id"])
```

`client.webhooks.verify` is the same function as `verify`.

## Usage

```python
from datetime import datetime, timezone

usage = client.usage.get()  # this month so far
# {cost, used: [{kind, unit, quantity, count, cost, unpriced}], messages: [{channel, received, sent}]}
october = client.usage.get(
    from_=datetime(2026, 10, 1, tzinfo=timezone.utc), to=datetime(2026, 11, 1, tzinfo=timezone.utc)
)
```

## Errors, retries and pages

Whatever the API refuses or fails raises one `ApiError`: `status` (HTTP), `code` to decide by and
`detail`, a sentence you can show.

```python
from signalcore import ApiError

try:
    client.agents.get("agt_…")
except ApiError as error:
    if error.code == "not_found":
        ...
```

| `status` | `code` | Means |
|---|---|---|
| 401 | `unauthorized` | no key, or not valid |
| 403 | `forbidden` | the key lacks the scope |
| 404 | `not_found` | it doesn't exist, or isn't yours |
| 409 | `conflict` | it clashes with something that exists |
| 422 | `invalid`, `busy`, `pending`, `files`, … | the request can't be done; a body of the wrong shape is `code == "error"` with the details in `detail` |
| 429 | `rate_limited` | more than 600 calls a minute for the key |

The SDK doesn't retry by itself. On `429`, wait and try again:

```python
import time


def with_retry(call, tries=5):
    for attempt in range(1, tries + 1):
        try:
            return call()
        except ApiError as error:
            if error.status != 429 or attempt == tries:
                raise
            time.sleep(2**attempt)


agents = with_retry(lambda: client.agents.list())
```

Every write takes `idempotency_key=`: sent again with the same key within 24 hours, it answers the
first answer and isn't done twice.

```python
client.conversations.say(conversation["id"], text="Hola", idempotency_key=f"msg-{message_id}")
client.orgs.create(name="Clínica Norte", idempotency_key=f"customer-{customer_id}")
```

Two lists come in pages, read whole with `iter`; every other list answers at once:

```python
for run in client.runs.iter(agent=agent["id"], status=["failed"]):
    print(run["id"])
for item in client.activity.iter(kind="conversation"):
    print(item)
```

## Every operation

Each operation of the [API reference](https://signalcore.ai/docs/referencia) is a method named
from its id, `<resource>_<verb>`: `agents_get` is `client.agents.get(agent_id)`,
`agents_memories_create` is `client.agents.memories_create(…)`, `templates_update_all` is
`client.templates.update_all(…)`. `signalcore.OPERATIONS` maps every id to its method. What is
deprecated says so.

- Path parameters are positional; the body's fields and the query are keyword arguments.
- A field left out isn't sent; `None` sends `null` (`signalcore.OMIT` leaves one out explicitly).
- Files go up as bytes with `content_type=` and come down as bytes.
- A tool's new name in `client.tools.edit` is `new_name=`.

## Developing it

Half of the package is generated from the API's contract (`v3/api/openapi.json`):
`_resources.py` and `_types.py`, by `python v3/clients/generate.py`. The rest is by hand:
`_base.py` (requests, errors, pages), `_extras.py` (the live socket, webhooks). When the contract
changes, generate again; CI fails while they differ.

```bash
uv sync
uv run pytest      # against the local API (v3-api) at http://localhost:8300, or SIGNALCORE_TEST_API
uv run mypy && uv run ruff check
```

The tests that need an API sign up an owner of their own there and skip themselves when none
answers.
