What is AI Agent Onboarding?
AI agent onboarding is the process of configuring, deploying, and activating an artificial intelligence agent for a new client — all within a compressed timeframe, ideally a single business day. Unlike traditional enterprise software onboarding that can drag on for weeks with endless configuration meetings, ticket queues, and manual provisioning, AI agent onboarding leverages automation, templated configurations, and pre-built integration layers to collapse the timeline dramatically.
At its core, the onboarding pipeline handles several critical tasks:
- Identity and access provisioning — creating client-specific credentials, API keys, and permission scopes
- Knowledge base injection — loading the client's documentation, FAQs, product catalogs, or internal wikis into the agent's retrieval system
- Tool and API binding — connecting the agent to the client's CRM, ticketing system, calendar, or database
- Brand and voice alignment — configuring tone, response style, escalation rules, and compliance guardrails
- Testing and validation — running a battery of simulated interactions to confirm the agent behaves correctly before going live
The goal is not just speed. It's reliability at speed. When a client signs up in the morning and their AI agent is answering customer inquiries by afternoon, that's the promise of a well-engineered onboarding system. The following tutorial walks through building such a system from scratch, with production-ready code and architectural decisions explained along the way.
Why One-Day Onboarding Matters
Time-to-value is the single most important metric in AI agent adoption. Research across SaaS platforms consistently shows that clients who reach their first meaningful interaction with an AI system within 24 hours convert at rates 3-5x higher than those who wait a week or more. Beyond conversion, fast onboarding directly impacts:
- Client confidence — a smooth, fast setup signals technical maturity and reduces buyer's remorse
- Support load reduction — the agent starts deflecting tickets immediately, often before the client's support team even realizes they needed help
- Iteration velocity — early feedback loops let you tune the agent while the client is still in the honeymoon phase of engagement
- Competitive differentiation — competitors who take two weeks to onboard will lose deals to your one-day promise
From an engineering perspective, one-day onboarding forces discipline. It requires you to eliminate manual steps that teams often tolerate for months: hand-editing YAML files, waiting for DNS propagation, manually approving OAuth scopes. Every step that cannot be fully automated becomes a bottleneck, and bottlenecks are simply not allowed in a one-day SLA. This constraint produces cleaner architecture, better test coverage, and more robust error handling — benefits that compound long after the onboarding sprint is over.
Architecture Overview
Before diving into code, let's establish the high-level architecture. The onboarding system consists of four primary services that work together in a directed pipeline:
┌─────────────────┐ ┌──────────────────┐ ┌─────────────────┐ ┌──────────────┐
│ Onboarding API │────▶│ Provisioning │────▶│ Agent Builder │────▶│ Validator │
│ (client input) │ │ Service │ │ Service │ │ Service │
└─────────────────┘ └──────────────────┘ └─────────────────┘ └──────────────┘
│ │ │ │
▼ ▼ ▼ ▼
Client config IAM + secrets RAG pipeline + Test suite
+ preferences + tenant setup tool bindings + go-live signal
Each service is independently deployable and communicates via gRPC with strict deadlines. The entire pipeline must complete within a p99 latency of 4 hours, leaving margin for client-facing communication and handoff. Let's build each piece.
Step 1: The Onboarding API — Capturing Client Intent
The onboarding API is the single entry point where client configuration enters the system. It accepts a structured payload describing the client's environment, use case, and preferences, then validates and normalizes the data before forwarding it downstream.
Here's a production-grade FastAPI endpoint that handles the initial onboarding request:
from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel, Field, field_validator
from typing import Optional, List, Dict
from enum import Enum
import uuid
import hashlib
from datetime import datetime
app = FastAPI()
class AgentUseCase(str, Enum):
CUSTOMER_SUPPORT = "customer_support"
SALES_ASSISTANT = "sales_assistant"
INTERNAL_KNOWLEDGE = "internal_knowledge"
TECHNICAL_TRIAGE = "technical_triage"
class IntegrationTarget(BaseModel):
provider: str = Field(..., description="e.g., zendesk, salesforce, jira, slack")
api_base_url: str
oauth_grant_id: Optional[str] = None
custom_fields: Dict[str, str] = Field(default_factory=dict)
class ClientOnboardingRequest(BaseModel):
organization_id: str = Field(..., min_length=3, max_length=64)
organization_name: str
agent_use_case: AgentUseCase
agent_display_name: str = Field(..., description="How the agent introduces itself")
tone_profile: str = Field(default="professional_friendly")
knowledge_sources: List[str] = Field(
...,
description="URLs or file paths to documentation, FAQs, product sheets"
)
integrations: List[IntegrationTarget] = Field(default_factory=list)
escalation_email: str
compliance_boundaries: Optional[List[str]] = Field(
default=None,
description="Topics the agent must refuse to discuss"
)
language: str = Field(default="en")
timezone: str = Field(default="UTC")
@field_validator('knowledge_sources')
@classmethod
def sources_must_be_reachable(cls, v):
if len(v) == 0:
raise ValueError('At least one knowledge source is required')
return v
class OnboardingReceipt(BaseModel):
onboarding_id: str
status: str
estimated_completion: str
pipeline_stages: Dict[str, str]
ONBOARDING_STORE = {} # In production, use Redis or PostgreSQL
@app.post("/v1/onboarding/initiate", response_model=OnboardingReceipt)
async def initiate_onboarding(
request: ClientOnboardingRequest,
background_tasks: BackgroundTasks
):
"""Initiates a one-day onboarding pipeline for a new client AI agent."""
onboarding_id = str(uuid.uuid4())
receipt_handle = hashlib.sha256(
f"{request.organization_id}:{datetime.utcnow().isoformat()}".encode()
).hexdigest()[:12]
# Persist the onboarding intent
ONBOARDING_STORE[onboarding_id] = {
"status": "provisioning",
"request": request.model_dump(),
"created_at": datetime.utcnow().isoformat(),
"stages": {
"provisioning": "pending",
"knowledge_ingestion": "pending",
"tool_binding": "pending",
"validation": "pending"
}
}
# Fire-and-forget the pipeline (in production, push to a message queue)
background_tasks.add_task(execute_onboarding_pipeline, onboarding_id)
return OnboardingReceipt(
onboarding_id=onboarding_id,
status="provisioning_started",
estimated_completion=f"{4 * 60} minutes from submission",
pipeline_stages=ONBOARDING_STORE[onboarding_id]["stages"]
)
This endpoint does three important things: it validates the client's intent with Pydantic's strict parsing, generates a unique onboarding identifier, and kicks off the background pipeline while immediately returning a receipt to the caller. The immediate response is crucial — it lets the client-facing dashboard show real-time progress rather than a spinning loader.
Step 2: Provisioning Service — IAM, Tenants, and Secrets
The provisioning service creates everything the agent needs to exist as a distinct, isolated tenant. This includes database schemas, API credentials, and identity bindings. The following service uses PostgreSQL for tenant data and HashiCorp Vault for secret management:
import asyncio
import asyncpg
import hvac
from typing import Dict, Any
class ProvisioningService:
"""Handles tenant creation, credential provisioning, and isolation setup."""
def __init__(self, db_dsn: str, vault_url: str, vault_token: str):
self.db_dsn = db_dsn
self.vault_client = hvac.Client(url=vault_url, token=vault_token)
async def provision_tenant(self, onboarding_id: str, org_id: str) -> Dict[str, Any]:
"""Creates an isolated tenant with dedicated schema and credentials."""
# Step 1: Create a PostgreSQL schema for tenant isolation
conn = await asyncpg.connect(self.db_dsn)
try:
schema_name = f"agent_{org_id.lower().replace('-', '_')}"
await conn.execute(f"""
CREATE SCHEMA IF NOT EXISTS {schema_name};
-- Set up conversation storage
CREATE TABLE IF NOT EXISTS {schema_name}.conversations (
conversation_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
session_id UUID NOT NULL,
user_id TEXT,
messages JSONB NOT NULL DEFAULT '[]',
created_at TIMESTAMPTZ DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW()
);
-- Set up analytics table
CREATE TABLE IF NOT EXISTS {schema_name}.analytics (
event_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
event_type TEXT NOT NULL,
payload JSONB,
recorded_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_conv_session
ON {schema_name}.conversations(session_id);
CREATE INDEX IF NOT EXISTS idx_analytics_type
ON {schema_name}.analytics(event_type);
""")
finally:
await conn.close()
# Step 2: Generate and store credentials in Vault
api_key = self._generate_secure_key(prefix="sk-agent-")
webhook_secret = self._generate_secure_key(prefix="whsec-")
vault_path = f"secret/agents/{org_id}"
self.vault_client.secrets.kv.v2.create_or_update_secret(
path=vault_path,
secret={
"api_key": api_key,
"webhook_secret": webhook_secret,
"schema_name": schema_name,
"created_at": str(asyncio.get_event_loop().time()),
"rotation_policy": "90_days"
}
)
return {
"schema_name": schema_name,
"api_key_hash": self._hash_for_logging(api_key),
"vault_path": vault_path,
"provisioned_at": str(asyncio.get_event_loop().time())
}
@staticmethod
def _generate_secure_key(prefix: str = "") -> str:
import secrets
return f"{prefix}{secrets.token_urlsafe(32)}"
@staticmethod
def _hash_for_logging(value: str) -> str:
import hashlib
return hashlib.sha256(value.encode()).hexdigest()[:8]
async def verify_tenant_health(self, org_id: str) -> bool:
"""Quick health check after provisioning."""
conn = await asyncpg.connect(self.db_dsn)
try:
schema_name = f"agent_{org_id.lower().replace('-', '_')}"
result = await conn.fetchval(f"""
SELECT EXISTS (
SELECT FROM information_schema.schemata
WHERE schema_name = $1
)
""", schema_name)
return bool(result)
finally:
await conn.close()
The provisioning service enforces tenant isolation at the database level using PostgreSQL schemas. Each client gets their own schema with dedicated conversation and analytics tables. Credentials are stored in Vault with a rotation policy attached — this is critical because onboarding creates long-lived secrets that must be managed from day one. The health check method lets downstream services verify provisioning succeeded before proceeding.
Step 3: Knowledge Base Injection — The RAG Pipeline
Knowledge ingestion is often the longest-running step in onboarding. The agent needs to understand the client's domain, and that means processing documentation, chunking it intelligently, generating embeddings, and loading them into a vector store. The following code implements a robust ingestion pipeline using LangChain-compatible primitives:
from typing import List, Dict, Optional
import asyncio
import hashlib
from datetime import datetime
import aiohttp
import tempfile
import os
class KnowledgeIngestionService:
"""
Ingests client knowledge sources, chunks documents, generates embeddings,
and loads everything into a tenant-isolated vector index.
"""
def __init__(
self,
vector_store_url: str,
embedding_model: str = "text-embedding-3-large",
chunk_size: int = 512,
chunk_overlap: int = 64
):
self.vector_store_url = vector_store_url
self.embedding_model = embedding_model
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
async def ingest_knowledge_sources(
self,
onboarding_id: str,
org_id: str,
sources: List[str],
language: str = "en"
) -> Dict[str, any]:
"""Main ingestion entry point. Processes all sources in parallel."""
ingestion_report = {
"onboarding_id": onboarding_id,
"total_sources": len(sources),
"total_chunks": 0,
"total_tokens": 0,
"source_details": [],
"started_at": datetime.utcnow().isoformat()
}
# Fetch all sources concurrently
async with aiohttp.ClientSession() as session:
fetch_tasks = [self._fetch_source(session, url) for url in sources]
source_contents = await asyncio.gather(*fetch_tasks, return_exceptions=True)
all_chunks = []
for i, content in enumerate(source_contents):
if isinstance(content, Exception):
ingestion_report["source_details"].append({
"url": sources[i],
"status": "failed",
"error": str(content)
})
continue
# Chunk the document
chunks = self._chunk_document(
content,
source_url=sources[i]
)
all_chunks.extend(chunks)
ingestion_report["source_details"].append({
"url": sources[i],
"status": "success",
"chunks_produced": len(chunks),
"characters": len(content)
})
ingestion_report["total_chunks"] = len(all_chunks)
# Generate embeddings and push to vector store
if all_chunks:
await self._embed_and_store(
org_id=org_id,
chunks=all_chunks,
language=language
)
ingestion_report["total_tokens"] = sum(
len(chunk["text"].split()) for chunk in all_chunks
)
ingestion_report["completed_at"] = datetime.utcnow().isoformat()
return ingestion_report
async def _fetch_source(
self,
session: aiohttp.ClientSession,
url: str
) -> str:
"""Fetches content from URL or local file path."""
if url.startswith("file://"):
path = url.replace("file://", "")
with open(path, "r", encoding="utf-8") as f:
return f.read()
async with session.get(url, timeout=aiohttp.ClientTimeout(total=120)) as resp:
resp.raise_for_status()
return await resp.text()
def _chunk_document(self, text: str, source_url: str) -> List[Dict]:
"""
Splits a document into overlapping chunks with metadata.
Uses a sliding window approach with configurable overlap.
"""
chunks = []
words = text.split()
for i in range(0, len(words), self.chunk_size - self.chunk_overlap):
chunk_words = words[i:i + self.chunk_size]
if not chunk_words:
break
chunk_text = " ".join(chunk_words)
chunk_id = hashlib.md5(
f"{source_url}:{i}".encode()
).hexdigest()
chunks.append({
"id": chunk_id,
"text": chunk_text,
"metadata": {
"source_url": source_url,
"chunk_index": i,
"word_count": len(chunk_words),
"ingested_at": datetime.utcnow().isoformat()
}
})
return chunks
async def _embed_and_store(
self,
org_id: str,
chunks: List[Dict],
language: str
):
"""
Generates embeddings and writes to tenant-isolated vector store.
In production, this would batch calls to OpenAI, Cohere, or a local model.
"""
# Simulated embedding generation — replace with actual API calls
import numpy as np
namespace = f"tenant_{org_id}_knowledge"
for chunk in chunks:
# In production: embedding = openai.Embedding.create(...)
# Here we simulate for demonstration
embedding_vector = np.random.randn(1536).tolist() # 1536-dim for text-embedding-3-large
# Store in vector DB with tenant namespace
await self._upsert_vector(
namespace=namespace,
vector_id=chunk["id"],
vector=embedding_vector,
metadata={
**chunk["metadata"],
"text_preview": chunk["text"][:200]
}
)
async def _upsert_vector(
self,
namespace: str,
vector_id: str,
vector: List[float],
metadata: Dict
):
"""
Upserts a single vector into the tenant's namespace.
In production, this hits Pinecone, Weaviate, pgvector, or Milvus.
"""
# Placeholder for actual vector DB upsert
# e.g., await pinecone_index.upsert(vectors=[...], namespace=namespace)
await asyncio.sleep(0.001) # Simulate IO
return True
The ingestion service processes all knowledge sources in parallel using asyncio.gather. Each source is fetched, chunked with a sliding window that respects overlap boundaries, and then embedded into a tenant-isolated vector namespace. The chunk size of 512 words with 64-word overlap is a good default for most support documentation — it keeps retrieval context focused without losing transitional meaning across chunk boundaries. The service returns a detailed ingestion report that feeds directly into the onboarding dashboard so the client can see exactly what was processed.
Step 4: Tool Binding — Connecting External Systems
An AI agent without tools is just a chatbot. Tool binding connects the agent to the client's real-world systems — their CRM, ticketing platform, calendar, or order management system. This step is where the agent gains the ability to act on behalf of the client. The binding service handles OAuth completion, API validation, and tool registration:
from typing import List, Dict, Optional
import asyncio
import json
from datetime import datetime
class ToolBindingService:
"""
Registers and validates external tool integrations for the AI agent.
Each tool binding includes OAuth verification, schema validation,
and a test invocation to confirm end-to-end connectivity.
"""
# Registry of supported integration providers
SUPPORTED_PROVIDERS = {
"zendesk": {
"tool_name": "ticket_lookup",
"description": "Search and retrieve Zendesk tickets by ID or status",
"required_scopes": ["read", "write:tickets"],
"test_endpoint": "/api/v2/tickets/count"
},
"salesforce": {
"tool_name": "crm_record_lookup",
"description": "Query Salesforce for accounts, contacts, and opportunities",
"required_scopes": ["api"],
"test_endpoint": "/services/data/v58.0/query?q=SELECT+COUNT(Id)+FROM+Account"
},
"slack": {
"tool_name": "send_slack_message",
"description": "Post messages to designated Slack channels",
"required_scopes": ["chat:write", "chat:read"],
"test_endpoint": "/api/conversations.list"
},
"jira": {
"tool_name": "jira_issue_ops",
"description": "Create, update, and query Jira issues",
"required_scopes": ["read:jira-work", "write:jira-work"],
"test_endpoint": "/rest/api/3/myself"
}
}
async def bind_tools(
self,
org_id: str,
integrations: List[Dict],
agent_id: str
) -> Dict[str, any]:
"""
Binds all requested integrations to the specified agent.
Validates each one before registering.
"""
binding_report = {
"agent_id": agent_id,
"total_requested": len(integrations),
"successful_bindings": 0,
"failed_bindings": 0,
"details": [],
"started_at": datetime.utcnow().isoformat()
}
binding_tasks = []
for integration in integrations:
provider = integration.get("provider", "").lower()
if provider not in self.SUPPORTED_PROVIDERS:
binding_report["details"].append({
"provider": provider,
"status": "skipped",
"reason": f"Provider '{provider}' is not in the supported registry"
})
continue
binding_tasks.append(
self._bind_single_tool(
org_id=org_id,
agent_id=agent_id,
provider=provider,
integration=integration
)
)
results = await asyncio.gather(*binding_tasks, return_exceptions=True)
for i, result in enumerate(results):
if isinstance(result, Exception):
binding_report["failed_bindings"] += 1
binding_report["details"].append({
"provider": list(self.SUPPORTED_PROVIDERS.keys())[i] if i < len(self.SUPPORTED_PROVIDERS) else "unknown",
"status": "failed",
"error": str(result)
})
else:
binding_report["successful_bindings"] += 1
binding_report["details"].append(result)
binding_report["completed_at"] = datetime.utcnow().isoformat()
return binding_report
async def _bind_single_tool(
self,
org_id: str,
agent_id: str,
provider: str,
integration: Dict
) -> Dict:
"""Validates and registers a single tool binding."""
provider_config = self.SUPPORTED_PROVIDERS[provider]
# Step 1: Verify OAuth grant is valid (in production, call the provider's token endpoint)
oauth_valid = await self._verify_oauth_grant(
provider=provider,
grant_id=integration.get("oauth_grant_id"),
required_scopes=provider_config["required_scopes"]
)
if not oauth_valid:
raise ValueError(f"OAuth grant for {provider} is invalid or missing required scopes")
# Step 2: Perform a connectivity test
api_base = integration.get("api_base_url", "")
test_result = await self._test_connectivity(
api_base_url=api_base,
test_endpoint=provider_config["test_endpoint"]
)
if not test_result["success"]:
raise ValueError(
f"Connectivity test for {provider} failed: {test_result.get('error')}"
)
# Step 3: Register the tool binding in the agent configuration store
tool_registration = {
"agent_id": agent_id,
"org_id": org_id,
"tool_name": provider_config["tool_name"],
"provider": provider,
"api_base_url": api_base,
"oauth_grant_id": integration.get("oauth_grant_id"),
"registered_at": datetime.utcnow().isoformat(),
"status": "active",
"test_verified": True
}
# In production, persist to a configuration database
# await db.tool_bindings.insert_one(tool_registration)
return {
"provider": provider,
"tool_name": provider_config["tool_name"],
"status": "bound",
"verified_at": datetime.utcnow().isoformat()
}
async def _verify_oauth_grant(
self,
provider: str,
grant_id: Optional[str],
required_scopes: List[str]
) -> bool:
"""
Verifies that the OAuth grant exists and has the required scopes.
In production, this calls the authorization server's introspection endpoint.
"""
if not grant_id:
return False
# Placeholder: In production, call provider's token introspection
# e.g., for Zendesk: GET /oauth/tokens/{grant_id}
await asyncio.sleep(0.001)
return True
async def _test_connectivity(
self,
api_base_url: str,
test_endpoint: str
) -> Dict:
"""
Performs a lightweight API call to verify connectivity.
Uses a short timeout to fail fast if the endpoint is unreachable.
"""
import aiohttp
try:
full_url = f"{api_base_url.rstrip('/')}{test_endpoint}"
async with aiohttp.ClientSession() as session:
async with session.get(
full_url,
timeout=aiohttp.ClientTimeout(total=10)
) as resp:
return {
"success": resp.status in [200, 201, 204],
"status_code": resp.status,
"error": None
}
except Exception as e:
return {
"success": False,
"status_code": None,
"error": str(e)
}
The tool binding service maintains a registry of supported providers with their required OAuth scopes and test endpoints. Each binding goes through three stages: OAuth scope verification, a connectivity smoke test, and persistent registration. The connectivity test is particularly important — it catches misconfigured base URLs, expired tokens, and firewall issues before the agent goes live. Failed bindings are reported individually so the client can fix specific integrations without blocking the entire onboarding.
Step 5: The Orchestrator — Putting the Pipeline Together
The orchestrator coordinates all four services, handles retries, updates progress state, and ultimately signals whether the agent is ready for production. This is the function referenced earlier as the background task in the onboarding API:
import asyncio
from datetime import datetime, timedelta
from enum import Enum
from typing import Dict, Any
class PipelineStage(str, Enum):
PROVISIONING = "provisioning"
KNOWLEDGE_INGESTION = "knowledge_ingestion"
TOOL_BINDING = "tool_binding"
VALIDATION = "validation"
COMPLETE = "complete"
FAILED = "failed"
PIPELINE_STATE: Dict[str, Dict[str, Any]] = {} # In production: Redis with TTL
async def execute_onboarding_pipeline(onboarding_id: str):
"""
Master orchestrator for the one-day onboarding pipeline.
Executes stages sequentially with retry logic and progress tracking.
"""
onboarding_data = ONBOARDING_STORE.get(onboarding_id)
if not onboarding_data:
raise ValueError(f"Onboarding {onboarding_id} not found")
request = onboarding_data["request"]
org_id = request["organization_id"]
# Initialize pipeline state
PIPELINE_STATE[onboarding_id] = {
"current_stage": PipelineStage.PROVISIONING,
"stage_history": [],
"started_at": datetime.utcnow().isoformat(),
"retry_counts": {}
}
# --- Stage 1: Provisioning ---
await _update_stage(onboarding_id, PipelineStage.PROVISIONING, "in_progress")
try:
provisioning = ProvisioningService(
db_dsn=os.getenv("POSTGRES_DSN", "postgresql://localhost:5432/agents"),
vault_url=os.getenv("VAULT_URL", "http://localhost:8200"),
vault_token=os.getenv("VAULT_TOKEN", "root-token")
)
provision_result = await provisioning.provision_tenant(onboarding_id, org_id)
health_ok = await provisioning.verify_tenant_health(org_id)
if not health_ok:
raise RuntimeError("Tenant health check failed after provisioning")
await _update_stage(onboarding_id, PipelineStage.PROVISIONING, "complete", provision_result)
except Exception as e:
await _update_stage(onboarding_id, PipelineStage.PROVISIONING, "failed", {"error": str(e)})
await _fail_pipeline(onboarding_id, str(e))
return
# --- Stage 2: Knowledge Ingestion ---
await _update_stage(onboarding_id, PipelineStage.KNOWLEDGE_INGESTION, "in_progress")
try:
ingestion = KnowledgeIngestionService(
vector_store_url=os.getenv("VECTOR_STORE_URL", "http://localhost:8000"),
chunk_size=512,
chunk_overlap=64
)
ingest_result = await ingestion.ingest_knowledge_sources(
onboarding_id=onboarding_id,
org_id=org_id,
sources=request["knowledge_sources"],
language=request.get("language", "en")
)
if ingest_result["total_chunks"] == 0:
raise RuntimeError("No knowledge chunks were produced — check source accessibility")
await _update_stage(onboarding_id, PipelineStage.KNOWLEDGE_INGESTION, "complete", ingest_result)
except Exception as e:
await _update_stage(onboarding_id, PipelineStage.KNOWLEDGE_INGESTION, "failed", {"error": str(e)})
await _fail_pipeline(onboarding_id, str(e))
return
# --- Stage 3: Tool Binding ---
await _update_stage(onboarding_id, PipelineStage.TOOL_BINDING, "in_progress")
try:
binding = ToolBindingService()
bind_result = await binding.bind_tools(
org_id=org_id,
integrations=request.get("integrations", []),
agent_id=f"agent-{org_id}"
)
await _update_stage(onboarding_id, PipelineStage.TOOL_BINDING, "complete", bind_result)
except Exception as e:
await _update_stage(onboarding_id, PipelineStage.TOOL_BINDING, "failed", {"error": str(e)})
await _fail_pipeline(onboarding_id, str(e))
return
# --- Stage 4: Validation ---
await _update_stage(onboarding_id, PipelineStage.VALIDATION, "in_progress")
try:
validation_result = await run_validation_suite(
onboarding_id=onboarding_id,
org_id=org_id,
agent_use_case=request["agent_use_case"],
tone_profile=request.get("tone_profile", "professional_friendly"),
compliance_boundaries=request.get("compliance_boundaries")
)
if not validation_result["passed"]:
raise RuntimeError(
f"Validation failed: {validation_result.get('failure_reasons', [])}"
)
await _update_stage(onboarding_id, PipelineStage.VALIDATION, "complete", validation_result)
except Exception as e:
await _update_stage(onboarding_id, PipelineStage.VALIDATION, "failed", {"error": str(e)})
await _fail_pipeline(onboarding_id, str(e))
return
# --- Pipeline Complete ---
await _update_stage(onboarding_id, PipelineStage.COMPLETE, "complete", {
"agent_ready": True,
"go_live_time": datetime.utcnow().isoformat(),
"total_duration_seconds": (
datetime.utcnow() - datetime.fromisoformat(PIPELINE_STATE[onboarding_id]["started_at"])
).total_seconds()
})
async def _update_stage(
onboarding_id: str,
stage: PipelineStage,
status: str,
details: Dict[str, Any] = None
):
"""Updates pipeline progress in the shared state store."""
PIPELINE_STATE[onboarding_id]["current_stage"] = stage
PIPELINE_STATE[onboarding_id]["stage_history"].append({
"stage": stage.value,
"status": status,
"timestamp": datetime.utcnow().isoformat(),
"details": details or {}
})
# Also update the API-visible store
if onboarding_id in ONBOARDING_STORE:
ONBOARDING_STORE[onboarding_id]["stages"][stage.value] = status
async def _fail_pipeline(onboarding_id: str, reason: str):
"""Marks the pipeline as failed with a reason."""
PIPELINE_STATE[onboarding_id]["current_stage"] = PipelineStage.FAILED
PIPELINE_STATE[onboarding_id]["failure_reason"] = reason
if onboarding_id in ONBOARDING_STORE:
ONBOARDING_STORE[onboarding_id]["status"] = "failed"
ONBOARDING_STORE[onboarding_id]["failure_reason"] = reason
The orchestrator runs each stage sequentially, but within each stage, operations are parallelized (sources fetched concurrently, tools bound concurrently). It maintains a detailed stage history that can be exposed via a progress endpoint for the client dashboard. Most importantly, it fails fast with specific error messages — a failed knowledge ingestion doesn't leave the system in a half-provisioned state because the orchestrator short-circuits and reports exactly what went wrong.
Step 6: Validation Suite — Confirming Agent Readiness
Validation is the final gate before the agent goes live. The validation suite runs a series of test conversations against the fully-assembled agent, checking for correct knowledge retrieval, appropriate tone, compliance with boundaries, and successful tool invocations:
import asyncio
from typing import List, Dict, Any
from datetime import datetime
# Pre-defined test scenarios for each use case
VALIDATION_SCENARIOS = {
"customer_support