TL;DR
- A verification instruction in the system prompt is not a gate; the model can enroll a contact it never verified, and nothing in the agent loop stops it.
- Scalekit's LangChain adapter catches every tool exception and returns f"Error executing tool {tool_name}: {str(e)}" as tool content, so a LeadIQ 401 or an exhausted credit balance arrives at the model as readable text rather than a failure signal.
- actions.langchain.get_tools(identifier=...) binds connected_account_id into each StructuredTool closure at bind time; hoisting that call to module scope makes every tenant execute against whichever tenant warmed the process first.
- LeadIQ authenticates with an API key and Outreach with OAuth 2.0, so the two credentials have different lifecycles but must resolve from one namespaced identifier per tenant user.
- The gate belongs in wrap_tool_call as deterministic code: leadiq_search_people_preview before spending a credit, leadiq_search_people before enrolling, and a short-circuit ToolMessage when the verdict is not PASS.
- Scalekit issues the credential per connected account and returns an execution_id per call, which gives the gate an audit record tying every verdict to the identity that produced it.
Three months after the outbound agent shipped, a rep opens Outreach and finds 340 prospects enrolled in a sequence with a 19% hard bounce rate. The agent's trace looks clean. It called leadiq_search_people, it called outreach_sequence_states_create, and the system prompt said, in bold, Never enroll a contact whose email has not been verified against LeadIQ. What the trace does not show is that on 340 of those runs, LeadIQ returned a plan error, the tool wrapper turned it into a sentence, and the model read that sentence as a result and moved on. Google calculates user-reported spam rate daily and expects bulk senders to stay under 0.1% and never reach 0.3%, per its email sender guidelines FAQ. The gate was in the prompt. Prompts are not gates.
Why a prompt-level verification gate cannot hold the line
Three separate mechanisms let an unverified contact through, and none of them are model quality problems.
The adapter converts failures into prose. The Scalekit Python SDK's LangChain adapter wraps each scoped tool in a callable whose except branch returns a string:
except Exception as e:
return f"Error executing tool {tool_name}: {str(e)}"
LeadIQ's prospect list tools return 401 from the prospector service on Freemium accounts, and credit-consuming searches fail once quota is gone. Both arrive as tool content. A ToolMessage carrying the word "Error" is still a ToolMessage with status="success", and the loop continues.
The adapter also stringifies successes. The same code path returns str(result_dict), a Python repr rather than JSON. Any gate that tries to parse the tool output the model saw is parsing a repr, not a contract.
create_agent gives you no hook at the tool boundary by default. LangChain's issue #33348 documents that create_agent hardcodes ToolNode instantiation with no handle_tool_errors parameter, and issue #33504 documents that invalid_tool_calls from malformed model output are dropped entirely rather than surfaced as recoverable errors. Separately, issue #2395 is the long-running report that agents jump to a conclusion when an observation is merely adjacent to the question.
Put together: the model sees a plausible sentence where a failure should be, has no structured error to react to, and has a strong prior toward completing the task it was given. The fix is not a firmer prompt. It is moving the decision out of the model's reach.
Recommended Reading: LangChain tool calling: how it works, where it stops, and how Scalekit completes it
The identity the gate runs under
A deterministic gate that runs under a shared credential is still the wrong gate. Verification reads a tenant's LeadIQ credits; enrollment writes into a rep's Outreach mailbox. Those are two different principals in two different auth dialects, and the gate's verdict is only meaningful if both resolve from the same run identity.
OAuth 2.0 (authorization code)
upsert_connected_account with the key
get_authorization_link then user consent
No expiry; rotates when the tenant rotates it
Access token refresh handled by Scalekit
Blast radius of the credential
Full LeadIQ account access
Scoped to the authorizing rep's Outreach permissions
What the agent is charged for
Nothing; writes count against the rep's mailbox
Failure signal when misconfigured
401 from the prospector service
Expired or revoked connected account
Two consequences follow. First, LeadIQ credits are a tenant-level resource being spent by an autonomous loop, which makes quota a security control and not just an ops metric. Second, execute_tool resolves a connected account by identifier plus connection_name, so the identifier must be namespaced per tenant. A bare rep_42 collides across customers; acme::rep_42 does not.
Recommended Reading: OAuth vs API keys for AI agents and how tool calling auth changes when you move from single-tenant to multi-tenant
Bind both connectors to one namespaced identifier
Create both connections once per environment in the Scalekit dashboard (AgentKit → Connections), note each Connection name, then attach credentials per user from your integrations settings page.
# connect.py
# Runs in your application, not in the agent process.
import os
from scalekit import ScalekitClient
scalekit = ScalekitClient(
env_url=os.environ["SCALEKIT_ENVIRONMENT_URL"], # AgentKit samples also use SCALEKIT_ENV_URL
client_id=os.environ["SCALEKIT_CLIENT_ID"],
client_secret=os.environ["SCALEKIT_CLIENT_SECRET"],
)
actions = scalekit.actions
def tenant_identifier(tenant_id: str, user_id: str) -> str:
"""Namespace the identifier. execute_tool resolves a connected account by
(identifier, connection_name), so an un-namespaced user id collides across tenants."""
return f"{tenant_id}::{user_id}"
def attach_leadiq(tenant_id: str, user_id: str, leadiq_api_key: str) -> str:
"""LeadIQ uses API key auth: there is no redirect flow and no authorization link.
The Python SDK maps authorization_details['static_auth'] onto a free-form
StaticAuth.details struct; 'username' is the field the LeadIQ connector reads."""
identifier = tenant_identifier(tenant_id, user_id)
actions.upsert_connected_account(
connection_name="leadiq", # must match the Connection name in the dashboard
identifier=identifier,
organization_id=tenant_id, # tags the account with its tenant for later listing
authorization_details={"static_auth": {"username": leadiq_api_key}},
)
return identifier
def start_outreach_oauth(tenant_id: str, user_id: str) -> str:
"""Outreach uses OAuth 2.0. Scalekit stores and refreshes the token; the key
never reaches the agent runtime and never enters LLM context."""
identifier = tenant_identifier(tenant_id, user_id)
link_response = actions.get_authorization_link(
connection_name="outreach",
identifier=identifier,
)
return link_response.link # redirect the rep here; verify on callback
The connection name is the single most common misconfiguration here. Scalekit generates suffixed names such as outreach-b1fqL2Dr when a connection is created more than once in an environment, and execute_tool will not fall back to a prefix match.
Resolve the authorized tool surface per run, not per process
The two catalogs together expose 66 tools. This agent needs nine. Binding the rest is an accuracy problem before it is a cost problem: the model picks leadiq_flat_advanced_search when it should pick leadiq_search_people_preview, and burns credits doing it. The agent sees only the tools the current user is authorized to call, not a connector catalog.
# tools.py
from scalekit import ScalekitClient
# The nine tools this agent is allowed to touch. Everything else stays out of context.
GATE_TOOLS = [
"leadiq_get_usage", # quota read, no credits
"leadiq_search_people_preview", # existence check, no credits
"leadiq_search_people", # authoritative work email, consumes credits
"leadiq_submit_person_feedback", # bounce reporting back to LeadIQ
"outreach_mailboxes_list", # mailbox_id is required to enroll
"outreach_sequences_list", # resolve sequence_id by name
"outreach_prospects_create", # create the prospect record
"outreach_sequence_states_create", # THE write being gated
"outreach_sequence_states_list", # read enrollment states, including bounced
]
def bind_tools(scalekit: ScalekitClient, identifier: str):
"""Returns LangChain StructuredTool objects scoped to one connected identity.
Do NOT hoist this to module scope. get_tools() calls list_scoped_tools() and
closes over the resolved connected_account_id inside each tool's callable.
A cached tool list executes against whichever identifier built it, which in a
multi-tenant worker means the first request to warm the process wins.
"""
return scalekit.actions.langchain.get_tools(
identifier=identifier,
tool_names=GATE_TOOLS,
page_size=50,
)
The tradeoff is real: building tools per run adds a list_scoped_tools round trip and a graph compile to every invocation. The alternative, a process-wide tool cache, trades that latency for cross-tenant execution under the wrong credential. Pay the round trip.
Recommended Reading: token-efficient tool calling and least privilege for agent tool calls
The verdict the model never writes
The gate's decision is a pure function over three inputs: the email the CRM proposes, LeadIQ's zero-credit existence check, and LeadIQ's authoritative work email. No model output participates.
Gate action on outreach_sequence_states_create
Provider email matches the proposed email after normalization
Provider email exists and differs from the proposed email
Allow, with the enrollment payload rewritten to the provider value
Preview returns no work email for the person
Short-circuit; no credit spent, no enrollment
Provider email domain differs from the account domain on record
Short-circuit; route to human review
Remaining LeadIQ credits below the run floor
Short-circuit; requeue the contact
# verdict.py
import re
from dataclasses import dataclass
from typing import Any, Iterable
EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
@dataclass(frozen=True)
class Verdict:
outcome: str # PASS | REPLACE | BLOCK_NO_DATA | BLOCK_DOMAIN | DEFER_QUOTA
email: str | None # the address the sequencer is allowed to use, if any
reason: str
def normalize(email: str) -> str:
"""Lowercase and trim only. Do not strip dots or plus-tags: those rules are
Gmail-specific and are wrong for corporate mailboxes, which is most of B2B."""
return (email or "").strip().lower()
def collect_work_emails(payload: Any) -> list[str]:
"""LeadIQ returns provider-shaped GraphQL payloads under ExecuteToolResponse.data.
Scalekit does not publish the searchPeople response schema, and hard-coding a
path is how gates silently start returning empty. Walk defensively, then pin the
real shape with a contract test against your own LeadIQ account."""
found: list[str] = []
def walk(node: Any, under_email_key: bool = False) -> None:
if isinstance(node, dict):
for key, value in node.items():
walk(value, under_email_key or "email" in key.lower())
elif isinstance(node, (list, tuple)):
for item in node:
walk(item, under_email_key)
elif isinstance(node, str) and under_email_key and EMAIL_RE.match(node.strip()):
found.append(normalize(node))
walk(payload)
return list(dict.fromkeys(found)) # dedupe, preserve order
def decide(
proposed_email: str,
account_domain: str,
preview_has_email: bool,
provider_emails: Iterable[str],
) -> Verdict:
if not preview_has_email:
return Verdict("BLOCK_NO_DATA", None, "LeadIQ preview reports no work email on file")
candidates = list(provider_emails)
if not candidates:
return Verdict("BLOCK_NO_DATA", None, "searchPeople returned no work email")
authoritative = candidates[0]
domain = authoritative.rsplit("@", 1)[-1]
if account_domain and domain != normalize(account_domain):
return Verdict("BLOCK_DOMAIN", None, f"provider domain {domain} != account {account_domain}")
if normalize(proposed_email) == authoritative:
return Verdict("PASS", authoritative, "provider email matches CRM record")
return Verdict("REPLACE", authoritative, f"CRM had {normalize(proposed_email)}")
The gate: a deterministic interceptor in wrap_tool_call
wrap_tool_call receives a ToolCallRequest and a handler, and you decide whether the handler runs at all. Calling it zero times is the short-circuit. Per-run state lives in the agent state rather than on the middleware instance, because a single middleware object is shared across concurrent runs in a worker.
# gate.py
import operator
from typing import Annotated, Any, Callable
from langchain.agents.middleware import AgentMiddleware, AgentState
from langchain.messages import ToolMessage
from langchain.tools.tool_node import ToolCallRequest
from langgraph.runtime import Runtime
from langgraph.types import Command
from typing_extensions import NotRequired
from verdict import Verdict, collect_work_emails, decide
CREDIT_TOOLS = {"leadiq_search_people"}
GATED_WRITE = "outreach_sequence_states_create"
class GateState(AgentState):
"""Custom state so per-run values are never shared between concurrent invocations."""
credits_available: NotRequired[int]
gate_verdicts: NotRequired[Annotated[list[dict[str, Any]], operator.add]]
class VerificationGate(AgentMiddleware):
"""Enforces two invariants the model cannot talk its way past:
1. No credit-consuming LeadIQ call without a zero-credit preview and quota headroom.
2. No Outreach sequence enrollment without a PASS or REPLACE verdict for that address.
"""
state_schema = GateState
def __init__(self, actions, identifier: str, credit_floor: int = 25):
super().__init__()
self.actions = actions # scalekit.actions, already bound to the environment
self.identifier = identifier # namespaced: "acme::rep_42"
self.credit_floor = credit_floor
def before_agent(self, state: GateState, runtime: Runtime) -> dict[str, Any] | None:
"""Read quota once per run. leadiq_get_usage takes no params and costs no credits."""
usage = self.actions.execute_tool(
tool_input={},
tool_name="leadiq_get_usage",
connection_name="leadiq",
identifier=self.identifier,
)
return {"credits_available": _extract_credits(usage.data)}
def wrap_tool_call(
self,
request: ToolCallRequest,
handler: Callable[[ToolCallRequest], ToolMessage | Command],
) -> ToolMessage | Command:
name = request.tool_call["name"]
args = request.tool_call["args"]
call_id = request.tool_call["id"]
# Invariant 1: protect the tenant's credit balance.
if name in CREDIT_TOOLS:
remaining = (request.state or {}).get("credits_available", 0)
if remaining < self.credit_floor:
return self._block(
call_id, name,
Verdict("DEFER_QUOTA", None, f"{remaining} credits left, floor is {self.credit_floor}"),
)
# Invariant 2: nothing enters a sequence without a fresh verdict.
if name == GATED_WRITE:
ctx = runtime_context(request)
verdict = self._verify(
proposed_email=ctx["proposed_email"],
account_domain=ctx["account_domain"],
linkedin_url=ctx.get("linkedin_url"),
full_name=ctx.get("full_name"),
company_name=ctx.get("company_name"),
)
if verdict.outcome in ("BLOCK_NO_DATA", "BLOCK_DOMAIN", "DEFER_QUOTA"):
return self._block(call_id, name, verdict)
# REPLACE rewrites the payload before the write executes. The model is
# never asked to correct itself, because asking is not enforcing.
if verdict.outcome == "REPLACE":
request = request.override(
tool_call={**request.tool_call, "args": {**args, "_verified_email": verdict.email}}
)
result = handler(request)
return Command(update={
"messages": [result],
"gate_verdicts": [{"tool": name, "outcome": verdict.outcome, "email": verdict.email}],
})
return handler(request)
def _verify(self, proposed_email, account_domain, linkedin_url, full_name, company_name) -> Verdict:
"""Both LeadIQ calls go through actions.execute_tool directly, not through the
model-facing StructuredTool. The adapter stringifies results and swallows
exceptions; the gate needs the typed response and the raised error."""
lookup = {k: v for k, v in {
"linkedin_url": linkedin_url,
"full_name": full_name,
"company_name": company_name,
}.items() if v}
preview = self.actions.execute_tool(
tool_input=lookup,
tool_name="leadiq_search_people_preview",
connection_name="leadiq",
identifier=self.identifier,
)
has_email = bool(
(preview.data or {}).get("data", {}).get("searchPeoplePreview", {}).get("hasEmail")
)
if not has_email:
return Verdict("BLOCK_NO_DATA", None, "preview reports no work email")
# Only now do we spend a credit.
person = self.actions.execute_tool(
tool_input={**lookup, "limit": 1, "contains_work_contact_info": True},
tool_name="leadiq_search_people",
connection_name="leadiq",
identifier=self.identifier,
)
return decide(
proposed_email=proposed_email,
account_domain=account_domain,
preview_has_email=True,
provider_emails=collect_work_emails(person.data),
)
@staticmethod
def _block(call_id: str, tool_name: str, verdict: Verdict) -> Command:
message = ToolMessage(
content=f"BLOCKED by verification gate: {verdict.outcome}. {verdict.reason}",
tool_call_id=call_id,
name=tool_name,
status="error", # a real error status, not a sentence that looks like one
)
return Command(update={
"messages": [message],
"gate_verdicts": [{"tool": tool_name, "outcome": verdict.outcome, "email": None}],
})
def runtime_context(request: ToolCallRequest) -> dict[str, Any]:
"""The contact under evaluation comes from your pipeline via runtime context,
never from the model's tool arguments. Model-authored input is not evidence."""
ctx = request.runtime.context
return {
"proposed_email": ctx.proposed_email,
"account_domain": ctx.account_domain,
"linkedin_url": ctx.linkedin_url,
"full_name": ctx.full_name,
"company_name": ctx.company_name,
}
def _extract_credits(payload: Any) -> int:
"""leadiq_get_usage returns plan credit counts and usage caps. Shapes vary by plan,
so fail closed at 0 rather than optimistically assuming headroom."""
def walk(node: Any) -> int | None:
if isinstance(node, dict):
for key, value in node.items():
if "available" in key.lower() and isinstance(value, (int, float)):
return int(value)
found = walk(value)
if found is not None:
return found
elif isinstance(node, list):
for item in node:
found = walk(item)
if found is not None:
return found
return None
return walk(payload) or 0
The critical line is status="error". It is the difference between the model reading a failure and the model reading a sentence.
Wiring the agent
The context dataclass carries the run's identity and the contact under evaluation. Middleware order matters: the first entry is the outermost wrapper.
# agent.py
import os
from dataclasses import dataclass
from langchain.agents import create_agent
from scalekit import ScalekitClient
from gate import GateState, VerificationGate
from tools import bind_tools
scalekit = ScalekitClient(
env_url=os.environ["SCALEKIT_ENVIRONMENT_URL"],
client_id=os.environ["SCALEKIT_CLIENT_ID"],
client_secret=os.environ["SCALEKIT_CLIENT_SECRET"],
)
SYSTEM_PROMPT = """You enroll a single verified contact into an Outreach sequence.
Steps:
1. Call outreach_sequences_list to resolve the sequence id by name.
2. Call outreach_mailboxes_list to resolve the sending mailbox id.
3. Call outreach_prospects_create for the contact.
4. Call outreach_sequence_states_create with prospect_id, sequence_id, mailbox_id.
A verification gate runs outside your control. If a tool returns BLOCKED, stop and
report the verdict. Do not retry, do not substitute an address, do not guess."""
@dataclass
class RunContext:
tenant_id: str
identifier: str
proposed_email: str
account_domain: str
full_name: str
company_name: str
linkedin_url: str | None
sequence_name: str
def run_once(ctx: RunContext) -> dict:
"""Built per run. See tools.bind_tools for why this is not cached across tenants."""
tools = bind_tools(scalekit, ctx.identifier)
agent = create_agent(
model="anthropic:claude-sonnet-4-6",
tools=tools,
system_prompt=SYSTEM_PROMPT,
state_schema=GateState,
context_schema=RunContext,
middleware=[
VerificationGate(
actions=scalekit.actions,
identifier=ctx.identifier,
credit_floor=25,
),
],
)
return agent.invoke(
{
"messages": [{
"role": "user",
"content": (
f"Enroll {ctx.full_name} at {ctx.company_name} "
f"into the '{ctx.sequence_name}' sequence."
),
}],
"gate_verdicts": [],
},
context=ctx,
)
if __name__ == "__main__":
result = run_once(RunContext(
tenant_id="acme",
identifier="acme::rep_42",
proposed_email="j.doe@acme-target.com",
account_domain="acme-target.com",
full_name="Jane Doe",
company_name="Acme Target",
linkedin_url="https://www.linkedin.com/in/janedoe",
sequence_name="Q3 Outbound - VP Eng",
))
print(result["gate_verdicts"])
Pair this with ToolCallLimitMiddleware if you want a hard ceiling on tool calls per run, independent of the credit floor.
Recommended Reading: human-in-the-loop tool calling for routing BLOCK_DOMAIN verdicts to an approval queue
Closing the loop: bounced states feed LeadIQ
Verification at enrollment time is a snapshot. Bounces are the ground truth that arrives later, and Outreach exposes them as sequence states. Reporting them back to LeadIQ improves the pool the next run reads from.
# reconcile.py
# Run on a schedule per tenant, outside the agent loop.
def reconcile_bounces(actions, identifier: str) -> int:
"""outreach_sequence_states_list supports filter_state values including
'bounced' and 'opted_out'. Each bounce is a data-quality signal LeadIQ accepts."""
states = actions.execute_tool(
tool_input={"filter_state": "bounced", "page_size": 200},
tool_name="outreach_sequence_states_list",
connection_name="outreach",
identifier=identifier,
)
reported = 0
for prospect_id, email in extract_bounced_prospects(states.data):
actions.execute_tool(
tool_input={
"value": email,
"status": "Invalid",
"type": "WorkEmail",
"invalid_reason": "EmailBounceCode550",
},
tool_name="leadiq_submit_person_feedback",
connection_name="leadiq",
identifier=identifier, # same namespaced identity, both connectors
)
reported += 1
return reported
Note the identifier: the same namespaced identity reads Outreach over OAuth and writes LeadIQ over an API key. That is what makes the loop auditable. Every execute_tool call returns an execution_id, and every connected account carries a connected_account_id; persisting the pair alongside each gate_verdicts entry gives you a per-action record of which identity produced which verdict.
Recommended Reading: audit trails for agent auth in B2B SaaS and agent tool observability: your agent is running — is it actually working?
What this gate does not do
Being precise about the boundary matters more than the gate looking comprehensive.
Address not in LeadIQ's database
leadiq_search_people_preview returns hasEmail: false
CRM address stale after a job change
Provider email differs, verdict is REPLACE
Right name, wrong company
Domain comparison produces BLOCK_DOMAIN
Mailbox deprovisioned yesterday
LeadIQ is a contact database, not an SMTP verifier
Catch-all domain accepting any local part
Requires SMTP-level probing, which LeadIQ does not expose
Needs a dedicated verification vendor in front of the sequencer
LeadIQ's own tool descriptions say leadiq_search_people returns verified work emails; that is a claim about provenance in their dataset, not a live deliverability check. If your bounce rate must sit under 2%, chain a dedicated verifier after this gate and before outreach_sequence_states_create. The gate's contribution is that it stops the class of bounce caused by the agent enrolling an address nobody checked, and it stops it deterministically.
FAQs
Can I skip the preview call and just run leadiq_search_people?
You can, and you will pay credits for contacts LeadIQ has no data on. leadiq_search_people_preview consumes no credits and returns hasEmail, which is exactly the cheap negative case. In a multi-tenant agent, the credits belong to the customer, which makes the preview step a spending control rather than an optimization.
Why call LeadIQ through actions.execute_tool inside the gate instead of the bound LangChain tool?
The LangChain adapter returns str(result_dict) on success and an error sentence on failure. The gate needs a typed ExecuteToolResponse with .data and .execution_id, and it needs exceptions to raise. Calling execute_tool directly gives both. The model-facing tools stay bound for the enrollment steps it actually drives.
Does LeadIQ need OAuth?
No. LeadIQ authenticates with an API key, so there is no authorization link and no redirect. Call upsert_connected_account with authorization_details={"static_auth": {"username": api_key}} and the tenant can call tools immediately. Scalekit stores the key and injects it at request time, so it never reaches the agent runtime or LLM context. Outreach in the same agent still uses the OAuth flow.
What happens on a LeadIQ Freemium account?
leadiq_get_lists, leadiq_get_list, leadiq_create_list, and leadiq_add_prospect_to_list return 401 from the prospector service. None of those four are in GATE_TOOLS, which is deliberate: keeping plan-gated tools out of the bound surface means the model never generates a call that turns into a 401-shaped sentence.
How do I stop one tenant's run from spending another tenant's credits?
Namespace the identifier, tag connected accounts with organization_id, and build the tool list inside the request. The first two make resolution unambiguous; the third prevents a cached StructuredTool closure, which holds a connected_account_id from whichever identifier created it, from executing under the wrong tenant.
Where does a human enter the loop?
BLOCK_DOMAIN is the verdict worth routing to a person, because it usually means the contact is real but attached to the wrong account. Wrap outreach_sequence_states_create with HumanInTheLoopMiddleware and interrupt only on that verdict rather than on every write.
Next steps to start building the email verification agent
- Create the leadiq and outreach connections in the Scalekit dashboard and record the exact connection names, including any generated suffix.
- Install the SDK with pip install scalekit-sdk-python and confirm leadiq_get_usage returns a credit balance for one test identifier.
- Write a contract test that pins the leadiq_search_people response shape for your plan, then replace collect_work_emails with the concrete path once it is stable.
- Ship VerificationGate with credit_floor set high and every verdict logged before you allow a single outreach_sequence_states_create through.
- Backfill reconcile_bounces against your last 30 days of Outreach sequence states to measure what the gate would have blocked.
- Browse the LeadIQ connector reference for the remaining nine tools, and the outbound prospecting agent template if you want the sourcing half of this pipeline wired the same way.
