Company Brain custom sources

A custom source puts your own system in the Company Brain: an internal wiki, a ticketing system, a document store behind your own API. Agents search it and read from it like any other source, with citations, scope, timeout, cache and the access log. An admin adds it in Platform > Agents > Knowledge; this page is for the engineers who build it. Your system decides what each person may see. Every call xpander makes to it carries a short-lived token, signed by your xpander install, that names the person who asked: their user id, email and groups. Your system verifies the token and answers only what that person may open, so an item it hides from someone never reaches them. There are two ways to connect a system:

	REST endpoint	MCP server
What your system exposes	`POST /search` and `POST /read` under a base URL (the contract)	A search function and a read function on an MCP connector
Where it is configured	Its address and authentication on the source	An MCP server already added in the Skills registry, with the authentication set there
Authentication	An API key in a header you choose, each person’s own OAuth sign-in, or none (the signed identity only)	Whatever the Skills registry entry uses: API key, custom headers, each person’s OAuth sign-in or their SSO token
Signed identity	On every call	On every call

The REST contract

Your system answers two POST requests under the base URL. Both take and return JSON. The full OpenAPI 3.1 description is below.

Search

POST {base}/search
X-Xpander-Identity: <signed identity token>
Authorization: Bearer <your API key>          (when the source uses an API key)
Content-Type: application/json

{"query": "what is our travel expense limit", "limit": 25, "scope": {"exclude": ["policies:drafts/"]}}

Field	Meaning
`query`	The person’s question as they asked it, up to 1000 characters
`limit`	The most items to return; xpander keeps the first 25
`scope`	Optional. The admin’s scope as a hint: `include` (only items under these ref prefixes or paths), `exclude` (never these) and `modified_after`. Honoring it saves work; xpander checks every item against the scope anyway

How admins scope a custom source by ref prefix and by path is under Custom source settings. Answer with the items this person may see, best match first:

{
  "items": [
    {
      "ref": "policies:travel",
      "title": "Travel and expenses policy",
      "snippet": "Meals are reimbursed up to $50 a day when traveling.",
      "url": "https://kb.example.com/policies/travel",
      "modified": "2026-09-01T10:00:00Z",
      "path": "/Policies/Finance"
    }
  ]
}

Field	Required	Meaning
`ref`	Yes	Your stable id for the item, up to 1000 characters. xpander passes it back to `/read`. Use `/` or `:` to nest refs (`wiki:eng/setup`) so admins can scope by ref prefix
`title`	Yes	Shown in the citation
`snippet`	No	The passage that answers the question; the answer quotes it. Up to 2000 characters
`url`	No	Where a person opens the item (an `http` or `https` link)
`modified`	No	ISO 8601 date. Needed when the source’s scope has an Older than cut-off: items without it are left out
`path`	No	A `/Folder/Sub` location. Needed when the scope has path rules: items without a path are left out by a path rule

Read

POST {base}/read
X-Xpander-Identity: <signed identity token>

{"ref": "policies:travel"}

{
  "title": "Travel and expenses policy",
  "text": "Meals are reimbursed up to $50 a day when traveling. Hotels: up to $250 a night.",
  "url": "https://kb.example.com/policies/travel",
  "modified": "2026-09-01T10:00:00Z"
}

text is plain text or Markdown; xpander keeps the first 200,000 characters. title, url, modified and path are optional.

Status codes

Your answer	What xpander does
`200`	Uses the answer
`401`	With Each person signs in (OAuth), asks that person to sign in again. Otherwise the source refused the configured key, and the person gets nothing from this source
`403`	This person may not use your system; they get nothing from it
`404` on `/read`	The item does not exist or this person may not open it. Answer the same way for both
Anything else, or no answer within the source’s timeout	The source is skipped for this question; the answer says it could not be reached

The identity token

Every call carries X-Xpander-Identity: a JSON Web Token signed with ES256 by your xpander install. Its header names the signing key (kid) and has typ: xpander-brain-identity+jwt.

Claim	Meaning
`sub`	The person’s xpander user id
`email`	Their email address
`groups`	The names of their xpander user groups
`group_ids`	The ids of the same groups
`org`	Your xpander organization id
`aud`	This source’s id, the Audience shown on the source. Reject any other
`iss`	The Issuer shown on the source
`iat`, `nbf`, `exp`	Issued, valid from, expires. A token lives at most 5 minutes
`jti`	A unique id for the token

The public keys are at the JWKS URL shown on the source: https://api.xpander.ai/.well-known/brain-identity-jwks.json on xpander cloud, and your own API host’s /.well-known/brain-identity-jwks.json on a self-hosted install. Cache the key set and fetch it again when a token names a kid you have not seen. Each install has its own signing key, and it is used for nothing but this token. Verify the signature, aud, exp and iss on every call, then apply your own permissions with sub, email and groups.

# pip install "pyjwt[crypto]"
import jwt

JWKS_URL = "https://api.xpander.ai/.well-known/brain-identity-jwks.json"
SOURCE_ID = "<the Audience shown on the source>"
ISSUER = "<the Issuer shown on the source>"

jwks = jwt.PyJWKClient(JWKS_URL, cache_keys=True, lifespan=300)


def person(token: str) -> dict:
    """The claims of a valid identity token; raises jwt.InvalidTokenError otherwise."""
    if jwt.get_unverified_header(token).get("typ") != "xpander-brain-identity+jwt":
        raise jwt.InvalidTokenError("not an xpander identity token")
    key = jwks.get_signing_key_from_jwt(token).key
    return jwt.decode(
        token, key, algorithms=["ES256"], audience=SOURCE_ID, issuer=ISSUER,
        leeway=30, options={"require": ["exp", "iat", "sub", "aud"]},
    )

# claims = person(request.headers["X-Xpander-Identity"])
# answer only what claims["sub"], claims["email"] and claims["groups"] may see

// npm install jose
import { createRemoteJWKSet, jwtVerify } from "jose";

const JWKS = createRemoteJWKSet(
  new URL("https://api.xpander.ai/.well-known/brain-identity-jwks.json"),
);
const SOURCE_ID = "<the Audience shown on the source>";
const ISSUER = "<the Issuer shown on the source>";

export async function person(token) {
  const { payload } = await jwtVerify(token, JWKS, {
    algorithms: ["ES256"],
    audience: SOURCE_ID,
    issuer: ISSUER,
    typ: "xpander-brain-identity+jwt",
    clockTolerance: 30,
    requiredClaims: ["exp", "iat", "sub"],
  });
  return payload; // payload.sub, payload.email, payload.groups
}

A minimal REST source

A complete source in Python (FastAPI) that verifies the token with person() above and filters by group:

# pip install fastapi uvicorn "pyjwt[crypto]"
from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel
import jwt

DOCS = [
    {"ref": "policies:travel", "title": "Travel and expenses policy", "allow": ["*"],
     "text": "Meals are reimbursed up to $50 a day when traveling."},
    {"ref": "finance:forecast/q4", "title": "Q4 revenue forecast", "allow": ["Finance"],
     "text": "The Q4 forecast is up 12% on Q3."},
]

app = FastAPI()


class Search(BaseModel):
    query: str
    limit: int = 25


class Read(BaseModel):
    ref: str


def caller(token: str) -> dict:
    try:
        return person(token)
    except jwt.InvalidTokenError as e:
        raise HTTPException(401, str(e))


def may_read(claims: dict, doc: dict) -> bool:
    return "*" in doc["allow"] or bool(set(doc["allow"]) & set(claims.get("groups") or []))


@app.post("/search")
def search(body: Search, x_xpander_identity: str = Header(...)):
    claims = caller(x_xpander_identity)
    words = body.query.lower().split()
    hits = [d for d in DOCS if may_read(claims, d) and any(w in (d["title"] + d["text"]).lower() for w in words)]
    return {"items": [{"ref": d["ref"], "title": d["title"], "snippet": d["text"]} for d in hits[: body.limit]]}


@app.post("/read")
def read(body: Read, x_xpander_identity: str = Header(...)):
    claims = caller(x_xpander_identity)
    doc = next((d for d in DOCS if d["ref"] == body.ref), None)
    if doc is None or not may_read(claims, doc):
        raise HTTPException(404, "not found")
    return {"title": doc["title"], "text": doc["text"]}

Run it behind TLS (uvicorn app:app behind your HTTPS proxy), use its base URL as the source’s Endpoint, and a person outside Finance never gets the forecast.

The MCP option

Pick an MCP server that is already in the Skills registry. xpander calls it over streamable HTTP: initialize, then tools/call on the Search tool with {"query": "...", "limit": 25, "scope": {...}} and on the Read tool with {"ref": "..."}. The functions answer in the same shapes as the REST contract: the search answer is {"items": [...]} and the read answer is {"title", "text", "url", "modified"}. xpander reads structuredContent when the answer has it, and otherwise parses the text content as JSON; a read may also answer plain text, which becomes the item’s text. A call that answers isError counts as a failure. The server’s authentication is the one set in the Skills registry: an API key, custom headers, each person’s OAuth sign-in (each person signs in once, from the source’s Connection tab or from Settings > Knowledge), or the person’s SSO token. Every call also carries X-Xpander-Identity, so the server can apply its own permissions even with a shared key. A server that supports only the older SSE transport cannot be used as a source.

Network rules

xpander calls a custom source from your install:

HTTPS only. On xpander cloud the endpoint must be reachable on the public internet; private, loopback and cloud metadata addresses are refused.
Self-hosted installs may call endpoints inside your network over HTTPS. Plain http:// is accepted only when the install sets BRAIN_CUSTOM_SOURCES_ALLOW_HTTP=true on agent-controller, and only for an address inside your network. Metadata addresses are never called.
No redirects to another host. A redirect to the same host with 307 or 308 is followed; any other redirect fails the call. Use the final URL as the endpoint.
Size and time. A search answer may be up to 2 MB and a read answer up to 5 MB. A call that takes longer than the source’s timeout is dropped for that question.
The calls do not go through an HTTPS_PROXY.

Air-gapped installations. Custom sources are the live sources an air-gapped installation can use: your system lives in your network, the JWKS is served by your own xpander API, and nothing leaves the installation. Add source offers Custom there.

Troubleshooting

A token that fails to verify on your side is usually one of: the JWKS fetched from another install’s API host, aud compared with something other than the source’s id, or a clock more than 30 seconds off. For the messages Test connection shows, see A custom source fails Test connection.

OpenAPI description

OpenAPI 3.1 for the REST contract

openapi: 3.1.0
info:
  title: xpander Company Brain custom source (REST contract)
  version: 1.0.0
paths:
  /search:
    post:
      operationId: search
      parameters:
        - $ref: "#/components/parameters/Identity"
      requestBody:
        required: true
        content:
          application/json:
            schema: { $ref: "#/components/schemas/SearchRequest" }
      responses:
        "200":
          description: The matching items this person may see, best first
          content:
            application/json:
              schema: { $ref: "#/components/schemas/SearchResponse" }
        "401": { description: The credential is missing or expired }
        "403": { description: This person may not search this system }
  /read:
    post:
      operationId: read
      parameters:
        - $ref: "#/components/parameters/Identity"
      requestBody:
        required: true
        content:
          application/json:
            schema: { $ref: "#/components/schemas/ReadRequest" }
      responses:
        "200":
          description: The item's text
          content:
            application/json:
              schema: { $ref: "#/components/schemas/ReadResponse" }
        "404": { description: No such item, or this person may not open it }
        "401": { description: The credential is missing or expired }
        "403": { description: This person may not read this system }
components:
  parameters:
    Identity:
      name: X-Xpander-Identity
      in: header
      required: true
      description: ES256 JWT naming the asking person; verify it against the JWKS on every call
      schema: { type: string }
  schemas:
    SearchRequest:
      type: object
      required: [query, limit]
      properties:
        query: { type: string, maxLength: 1000 }
        limit: { type: integer, minimum: 1, maximum: 100 }
        scope:
          type: object
          properties:
            include: { type: array, items: { type: string } }
            exclude: { type: array, items: { type: string } }
            modified_after: { type: string, format: date-time }
    SearchResponse:
      type: object
      required: [items]
      properties:
        items:
          type: array
          maxItems: 100
          items: { $ref: "#/components/schemas/SearchItem" }
    SearchItem:
      type: object
      required: [ref, title]
      properties:
        ref: { type: string, maxLength: 1000 }
        title: { type: string, maxLength: 500 }
        snippet: { type: string, maxLength: 2000 }
        url: { type: string, format: uri }
        modified: { type: string, format: date-time }
        path: { type: string, maxLength: 2000 }
    ReadRequest:
      type: object
      required: [ref]
      properties:
        ref: { type: string, maxLength: 1000 }
    ReadResponse:
      type: object
      required: [text]
      properties:
        title: { type: string }
        text: { type: string }
        url: { type: string, format: uri }
        modified: { type: string, format: date-time }
        path: { type: string }

Overview

REST API - Conversations

REST API - Agents

REST API - Tasks

Agent Workspace

REST API - Knowledge

REST API - Skills

REST API - Workflows

REST API - Custom Functions

REST API - Scheduled Tasks

REST API - Self-Schedules

REST API - Misc

Channels and webhooks

Advanced

Company Brain custom sources

The REST contract

Search

Read

Status codes

The identity token

A minimal REST source

The MCP option

Network rules

Troubleshooting

OpenAPI description

​The REST contract

​Search

​Read

​Status codes

​The identity token

​A minimal REST source

​The MCP option

​Network rules

​Troubleshooting

​OpenAPI description

The REST contract

Search

Read

Status codes

The identity token

A minimal REST source

The MCP option

Network rules

Troubleshooting

OpenAPI description