Custom Endpoints

Custom endpoints are a first-class building block in Agentomatic — right alongside agents, plugins, and pipelines. An endpoint is a user-defined HTTP API that typically calls one or more deployed model services (or any HTTP API) via authenticated httpx requests, aggregates their outputs, and exposes the result as a clean REST route.

They are designed for the common production need: "call my deployed models (with auth), combine the results, and feed them into an agent or pipeline" — with minimal code.

Endpoints give you:

• Auto-discovery from the endpoints/ directory
• Auto-generated /{path}, /health, and /info routes
• Strict Pydantic request/response validation
• Built-in authentication to upstreams (API key, bearer, basic, OAuth2)
• Fan-out to multiple upstreams with aggregation strategies
• Retries, timeouts, and Prometheus metrics out of the box
• Direct use as a pipeline step

How it works

flowchart LR
    Client([Client]) --> EP[Custom Endpoint]
    EP -->|httpx + auth| M1[Model Service A]
    EP -->|httpx + auth| M2[Model Service B]
    EP -->|httpx + auth| M3[Model Service C]
    M1 --> AGG{Aggregate}
    M2 --> AGG
    M3 --> AGG
    AGG --> EP
    EP --> Client

1. Place each endpoint in the endpoints/ directory (its own folder or module).
2. Subclass BaseEndpoint, declare upstreams, and pick an aggregation.
3. The platform discovers it and mounts routes under /api/v1/endpoints/{endpoint_name}.

Scaffolding an endpoint

agentomatic init ensemble --template endpoint --dir endpoints

This creates:

endpoints/ensemble/
├── __init__.py       # Package marker
├── endpoint.py       # BaseEndpoint subclass with upstreams
├── .env.example      # Upstream URLs + credentials
└── README.md         # Usage docs

Minimal example

The default handle fans out to every configured upstream — so you often only need configuration, no imperative code:

from agentomatic.endpoints import (
    AggregationStrategy,
    AuthType,
    BaseEndpoint,
    UpstreamAuthConfig,
    UpstreamConfig,
)


class EnsembleEndpoint(BaseEndpoint):
    endpoint_name = "ensemble"
    endpoint_description = "Aggregate predictions from deployed models."
    path = "/predict"
    aggregation = AggregationStrategy.MAJORITY

    upstreams = [
        UpstreamConfig(
            name="model_a",
            base_url="${MODEL_A_URL}",
            path="/v1/predict",
            auth=UpstreamAuthConfig(type=AuthType.BEARER, api_key="${MODEL_A_TOKEN}"),
        ),
        UpstreamConfig(
            name="model_b",
            base_url="${MODEL_B_URL}",
            path="/v1/predict",
            auth=UpstreamAuthConfig(
                type=AuthType.OAUTH2_CLIENT_CREDENTIALS,
                token_url="${MODEL_B_TOKEN_URL}",
                client_id="${MODEL_B_CLIENT_ID}",
                client_secret="${MODEL_B_CLIENT_SECRET}",
            ),
        ),
    ]

Mounted routes:

• POST /api/v1/endpoints/ensemble/predict — invoke the endpoint
• GET /api/v1/endpoints/ensemble/health — readiness
• GET /api/v1/endpoints/ensemble/info — metadata
• GET /api/v1/endpoints — list all endpoints

Secrets stay out of code

Any string field supports ${ENV} interpolation, resolved lazily at request time and never persisted or logged.

Authentication to upstreams

Each upstream configures its own auth, so different models can authenticate differently.

API keyBearerBasicOAuth2 (client credentials)

UpstreamAuthConfig(
    type=AuthType.API_KEY,
    api_key="${SERVICE_KEY}",
    header_name="X-API-Key",
    header_prefix="",  # no prefix
)

UpstreamAuthConfig(type=AuthType.BEARER, api_key="${SERVICE_TOKEN}")

UpstreamAuthConfig(
    type=AuthType.BASIC,
    username="${SERVICE_USER}",
    password="${SERVICE_PASSWORD}",
)

UpstreamAuthConfig(
    type=AuthType.OAUTH2_CLIENT_CREDENTIALS,
    token_url="${TOKEN_URL}",
    client_id="${CLIENT_ID}",
    client_secret="${CLIENT_SECRET}",
    scope="predict:read",       # optional
    audience="${AUDIENCE}",     # optional
)

The access token is fetched from token_url and cached until shortly before expiry (controlled by token_leeway).

Aggregation strategies

When an endpoint calls multiple upstreams, aggregation controls how their responses combine:

Strategy	Behaviour
ALL	Return a mapping of {upstream_name: data}; succeeds only if all succeed.
FIRST_SUCCESS	Return the first successful response.
MAJORITY	Return the most common response (weighted by UpstreamConfig.weight).

Typed schemas

Provide typed request/response schemas via generics for full OpenAPI docs:

from pydantic import BaseModel
from agentomatic.endpoints import BaseEndpoint


class ScoreRequest(BaseModel):
    text: str


class ScoreResponse(BaseModel):
    label: str
    confidence: float


class ScorerEndpoint(BaseEndpoint[ScoreRequest, ScoreResponse]):
    endpoint_name = "scorer"
    path = "/score"

    async def handle(self, request: ScoreRequest) -> ScoreResponse:
        ok, aggregated, results = await self.models.fan_out(request.model_dump())
        return ScoreResponse(label=aggregated["label"], confidence=aggregated["score"])

HTTP methods and browser-safe input

POST is the default endpoint method and receives the typed request as JSON. You may expose read-only endpoint logic with GET or HEAD; their typed input is automatically published as query parameters, not a request body. This matters for browser clients, which are forbidden from sending a body with a GET or HEAD request.

class LookupEndpoint(BaseEndpoint[ScoreRequest, ScoreResponse]):
    endpoint_name = "lookup"
    path = "/score"
    methods = ["GET", "POST"]

    async def handle(self, request: ScoreRequest) -> ScoreResponse:
        return ScoreResponse(label=request.text, confidence=1.0)

Both operations use the same Pydantic contract:

# GET / HEAD use URL query parameters
curl "http://localhost:8000/api/v1/endpoints/lookup/score?text=hello"

# POST (and other write methods) use a JSON body
curl -X POST http://localhost:8000/api/v1/endpoints/lookup/score \
  -H "Content-Type: application/json" -d '{"text":"hello"}'

For query methods, repeat an array field (?tag=red&tag=blue) and encode an object-valued field as JSON (?filters={"region":"eu"}; URL-encode it in real clients). Agentomatic validates both forms against the same input model.

The Studio Endpoints page reads each registered method directly from OpenAPI. If an endpoint exposes more than one method, choose the method in the card; its input form and output summary update to that operation's live schema. The Pipeline Builder and Task Board use the same discovery flow when an endpoint is selected, so a GET-only endpoint still exposes its real input and output fields for mapping and independent testing.

Custom logic

Override handle for anything beyond fan-out — call upstreams conditionally, post-process, merge with local computation, etc. The self.models client exposes fan_out(...) and per-upstream get(name).request(...).

Using endpoints in pipelines

Endpoints are usable as a pipeline step, so you can fetch data from your deployed models and pass it straight to an agent:

name: enrich_then_answer
steps:
  - endpoint: ensemble
    name: predictions
    input:
      payload: $.input.*
  - agent: answerer
    name: answer
    input:
      current_query: $.input.query
      predictions: $.steps.predictions.aggregated

See the Pipelines guide for the full step reference.

Registering programmatically

You can also register endpoints without discovery:

from agentomatic import AgentPlatform

platform = AgentPlatform(endpoints_dir="endpoints")
platform.register_endpoint(EnsembleEndpoint())
app = platform.build()

Observability

Every endpoint call emits Prometheus metrics (agentomatic_endpoint_calls_total, agentomatic_endpoint_duration_seconds), and every upstream call emits agentomatic_upstream_* metrics. See the Observability guide for the ready-to-run Grafana stack.