Integrations

External models, LangChain, and notifications

Connect external models or use Adaptive with LangChain.

External models

Connect proprietary models (OpenAI, Azure, Google, Anthropic, NVIDIA NIM) to use them through the Adaptive API with interaction and metrics logging.

OpenAI (direct)

external_model = adaptive.models.add_external(
    provider="open_ai",
    external_model_id="GPT4O",
    name="GPT-4o",
    api_key="OPENAI_API_KEY"
)

Supported model IDs: GPT4O, GPT4O_MINI, GPT4, GPT4_TURBO, GPT3_5_TURBO

Azure OpenAI

external_model = adaptive.models.add_external(
    provider="azure",
    external_model_id="DEPLOYMENT_NAME",
    name="Azure GPT-4o",
    api_key="AZURE_API_KEY",
    endpoint="https://aoairesource.openai.azure.com"
)

The external_model_id is your deployment name, and endpoint is your Azure OpenAI subscription endpoint.

Google

external_model = adaptive.models.add_external(
    provider="google",
    external_model_id="gemini-1.5-pro",
    name="Gemini 1.5 Pro",
    api_key="GOOGLE_API_KEY"
)

Supported models: Gemini model variations (excluding embeddings).

Anthropic

external_model = adaptive.models.add_external(
    provider="anthropic",
    external_model_id="claude-sonnet-4-5-20250929",
    name="Claude Sonnet 4.5",
    api_key="ANTHROPIC_API_KEY"
)

Use the model ID from Anthropic’s API documentation as the external_model_id.

Once connected, attach the model to a project and make inference requests like any other model.

LangChain

Adaptive is compatible with LangChain chat model classes.

ChatOpenAI

from langchain_openai import ChatOpenAI
import os

os.environ["OPENAI_API_KEY"] = "ADAPTIVE_API_KEY"

llm = ChatOpenAI(
    model="project_key/model_key",  # model_key is optional
    base_url="ADAPTIVE_URL/api/v1",
)

messages = [
    ("system", "You are a helpful assistant that translates English to French."),
    ("human", "I love programming."),
]
response = llm.invoke(messages)

ChatGoogleGenerativeAI

from langchain_google_genai import ChatGoogleGenerativeAI
import os

os.environ["GOOGLE_API_KEY"] = "ADAPTIVE_API_KEY"

llm = ChatGoogleGenerativeAI(
    model="project_key",
    client_options={"api_endpoint": "ADAPTIVE_URL/api/v1/google"},
    transport="rest",
)

messages = [
    ("system", "You are a helpful assistant that translates English to French."),
    ("human", "I love programming."),
]
response = llm.invoke(messages)

Platform notifications

Subscribe to job events and receive alerts through Slack, email, or webhook.

Topic patterns

Topics follow the format project:<project_id>:job:<job_id>:<event_type>. Subscribe using patterns with wildcard support:

Pattern	Matches
project:*:job:*:completed	Any job completion in any project
project:*:job:*:failed	Failed jobs across all projects
project:*:job:*:running	Job start events across all projects
project:*:job:**	All job events (any status) in any project

Use * to match a single segment and ** to match zero or more segments.

Event types: running, completed, failed, cancelled.

Integration types

Three delivery methods are available:

Type	Setup	Details
Slack	Webhook URL, optional bot token	Posts to a channel via incoming webhook
Email (SMTP)	SMTP host, port, credentials, from/to addresses	Sends to specified recipients
Webhook	HTTP endpoint URL, method, custom headers	HTTP request with event payload

Each integration can subscribe to different topic patterns. You can create multiple integrations of the same type for different channels or recipients.

SDK methods

Create

from adaptive_sdk.input_types import (
    ConnectionConfigInputSlack,
    ConnectionConfigInputSmtp,
    ConnectionConfigInputWebhook,
    SubscriptionInput,
    DeliveryScopeInput,
)

# Slack integration
integration = adaptive.integrations.create(
    team="my-team",
    name="Job alerts",
    provider="slack",
    connection=ConnectionConfigInputSlack(
        webhook_url="https://hooks.slack.com/services/T.../B.../xxx",
    ),
    subscriptions=[
        SubscriptionInput(
            topic_pattern="project:*:job:*:completed",
            scope=DeliveryScopeInput.TEAM,
        ),
        SubscriptionInput(
            topic_pattern="project:*:job:*:failed",
            scope=DeliveryScopeInput.TEAM,
        ),
    ],
)

# SMTP (email) integration
integration = adaptive.integrations.create(
    team="my-team",
    name="Failure alerts",
    provider="smtp",
    connection=ConnectionConfigInputSmtp(
        host="smtp.example.com",
        port=587,
        username="[e-mail]",
        password="SMTP_PASSWORD",
        from_email="[e-mail]",
        to_emails=["[e-mail]"],
    ),
    subscriptions=[
        SubscriptionInput(
            topic_pattern="project:*:job:*:failed",
            scope=DeliveryScopeInput.TEAM,
        ),
    ],
)

# Webhook integration
integration = adaptive.integrations.create(
    team="my-team",
    name="Job webhook",
    provider="webhook",
    connection=ConnectionConfigInputWebhook(
        url="https://example.com/hooks/adaptive",
        method="POST",
        headers={"Authorization": "Bearer WEBHOOK_SECRET"},
    ),
    subscriptions=[
        SubscriptionInput(
            topic_pattern="project:*:job:**",
            scope=DeliveryScopeInput.TEAM,
        ),
    ],
)

Delivery policy

Control duplicate notification behavior with the delivery_policy parameter:

Policy	Behavior
multishot	Send a notification every time (default)
singleshot	Send only once per event

adaptive.integrations.create(
    team="my-team",
    name="One-shot alerts",
    provider="slack",
    connection=ConnectionConfigInputSlack(
        webhook_url="https://hooks.slack.com/services/T.../B.../xxx",
    ),
    subscriptions=[
        SubscriptionInput(
            topic_pattern="project:*:job:*:completed",
            scope=DeliveryScopeInput.TEAM,
        ),
    ],
    delivery_policy="singleshot",
)

Delivery scopes

Subscriptions include a scope that controls who the notification targets:

Scope	Description
DeliveryScopeInput.USER	Deliver to specific users
DeliveryScopeInput.TEAM	Deliver to the team
DeliveryScopeInput.ORGANIZATION	Deliver to the organization
DeliveryScopeInput.ADMIN	Deliver to admins

Manage

# List integrations for a team
integrations = adaptive.integrations.list(team="my-team")

# Get a specific integration
integration = adaptive.integrations.get(id="INTEGRATION_ID")

# Disable an integration
adaptive.integrations.update(id="INTEGRATION_ID", enabled=False)

# Update subscriptions
adaptive.integrations.update(
    id="INTEGRATION_ID",
    subscriptions=[
        SubscriptionInput(
            topic_pattern="project:*:job:*:completed",
            scope=DeliveryScopeInput.TEAM,
        ),
    ],
)

# Delete an integration
adaptive.integrations.delete(id="INTEGRATION_ID")

Test

Send a test notification to verify your integration is configured correctly.

from adaptive_sdk.input_types import NotificationPayload, JobUpdatePayload

result = adaptive.integrations.test_notification(
    topic="project:my-project-id:job:test-job-id:completed",
    payload=NotificationPayload(
        job_update=JobUpdatePayload(
            job_id="test-job-id",
            job_name="test-job",
            job_status="Done",
            job_progress=100.0,
            job_started_at="2025-01-01T00:00:00Z",
            project_id="my-project-id",
            job_input={},
            job_outputs=[],
        ),
    ),
    scope_team="my-team",
)

List providers

Query available integration providers and their configuration details.

providers = adaptive.integrations.list_providers()

# Get details for a specific provider
slack_provider = adaptive.integrations.get_provider(name="slack")

Permissions

Creating and managing integrations requires integration:create, integration:read, integration:update, or integration:delete permissions. See Permissions for role configuration.