TL;DR
- A Lusha lookalike expansion run is not one call. It is lushamcp_lookalike_companies off a seed set, lushamcp_prospecting_company_search_by_text to widen, lushamcp_prospecting_company_enrich on the shortlist, lushamcp_signal_score_companies to rank, and two Tables calls to persist. Five of those six bill credits, and three of them bill per result returned.
- The Lusha credential is an API key created at the account level. It names an account, not a person, so quota and identity both detach from the user: rate limits and monthly credit caps are enforced per key, and every Tables route needs an email the credential cannot supply.
- That email is a tool parameter. Scalekit's connector schema states it "may still be passed to act on behalf of another owner," which makes list ownership model-controlled unless you strip it from the input_schema and inject it in the handler closure.
- dedupeSessionId looks like a pagination token and behaves like a spend multiplier: every "get more lookalikes" page bills per result returned. Page limits belong in a PreToolUse hook, not in a system prompt.
- Four of the connector's 42 tools (lushamcp_recommendations_*) are OAuth-only and return 403 on an API-key session, and five more carry a STALE marker as of 2026-08-19. Registering the full surface guarantees the model selects tools that cannot work.
- actions.tools.list_scoped_tools cuts 42 tools to the seven this run needs, and execute_tool resolves the rep's vaulted key server-side on every call, so no Lusha key enters the process or the model context. Budget then becomes a scope you can enforce, and every credit lands on a named identifier in the audit log.
A rep pastes one Lusha API key into .env, wires it into a Claude Agent SDK tool, and the expansion agent works. It seeds off the ten best-fit closed-won accounts, pulls 25 lookalikes, enriches them, scores them against active hiring and headcount signals, and writes a ranked table back into Lusha. Green run. The rep gets a list they act on.
Then the second rep runs it on Monday.
The table lands in the first rep's Lusha workspace, because email is a parameter and the model filled it with the only address in its context. The month's credit cap on the key is gone by Tuesday, spent on a dedupeSessionId loop chasing "more lookalikes" that nobody bounded. And every other rep's run now fails with a 402, which surfaces to them not as an error but as an expansion list that is quietly, plausibly short.
None of that is a reasoning failure. It is three properties of the Lusha connector colliding with the Agent SDK's single-user defaults.
What One Expansion Run Actually Costs on the Lusha Connector
The Lusha MCP connector exposes 42 tools. None of them takes "find me accounts like my best customers" and returns a ranked list; a reasoning loop has to sequence narrow, individually priced entry points. Per Lusha's credit billing reference, the price is attached to results, not to calls.
5 requests per minute upstream
lushamcp_lookalike_companies
seeds 5-100 identifiers, exclude 500, limit 1-100
lushamcp_prospecting_company_search_by_text
text 2-1024 chars; no page-size parameter
lushamcp_prospecting_company_enrich
Per result, plus per revealed field
lushamcp_signal_score_companies
Per matched signal per result
lushamcp_table_create, lushamcp_table_add_entities
Create is free; adding companies bills per newly-added company
Four properties of that table drive every design decision below.
- The expensive step is the one the model controls. limit on lookalike_companies defaults to 25 and accepts 100. Nothing in the schema stops a loop from asking for 100 four times.
- The widen step has no size dial. prospecting_company_search_by_text accepts text and an opaque pagination_token, nothing else. Page count is the only lever you own.
- Reads are billable, so the agent must not check its own work. lushamcp_table_get_entities bills per row returned. It is absent from this build on purpose.
- reveal is priced per field, per company, and validated upstream. The connector schema names the source of truth: the prior search preview's canReveal[].field, with per-field cost in canReveal[].credits. Omit reveal on company enrich and V3 returns only the free intent field; values absent from that preview are rejected by V3.
Five tools carry a STALE marker in the connector docs as of 2026-08-19, including lushamcp_prospecting_company_search (renamed to the _by_text variant) and both prospecting_*_filters resolvers, which have no replacement. That second loss matters: the structured, enumerable filter surface is gone from the MCP tool list, so the targeting criteria are now a natural-language string Lusha converts to filters server-side. The auditable artifact for "why these accounts" becomes the exact text you sent plus the model's reason_for_invocation, and nothing else.
Recommended reading: OAuth vs API Keys for AI Agents covers why an API-key connector needs more infrastructure around it, not less; the same credential model applies either way.
The Lusha Credential Names an Account, Not a Person
This is where Lusha diverges from a connector you have integrated over OAuth. Per Lusha's API reference, the key is available to Admins and Managers on Premium and Scale plans, it is shared at the account level, and an account can hold several. There is no user inside it.
Three planes fall out of that, and they fail in three different ways.
The Lusha API key in the vault
Which Lusha account the call runs against
401 on an invalid key, 403 when the account is inactive or the feature is off-plan
Per-key rate limit and per-key monthly credit cap
429 on rate or daily quota, 402 on insufficient credits
email on every Tables route
Which user in that account owns the table
Nothing; the list lands in the wrong workspace
The third row is the dangerous one, and it is the one a reasoning loop can reach.
- The key is the unit of quota, not the user. Lusha's help centre documents the exact symptom: calls rejected with a credit-limit error while the dashboard still shows credits remaining, because the cap is set per key. One shared key means one rep's pagination loop rate-limits and out-spends every other rep, and the victim sees a short list rather than a failure.
- Ownership is an explicit act-on-behalf-of parameter. The connector schema for lushamcp_table_list states email is required when authenticating with an API key, "there is no signed-in user to default to." On the write routes it reads: "it may still be passed to act on behalf of another owner." That is a documented delegation mechanism sitting in a JSON Schema the model fills in.
- Tenant isolation is entirely yours. Two customers means two Lusha accounts and two keys. Nothing in the key, the tool call, or the response distinguishes them. The connected account and its identifier are the only boundary.
Access Control for Multi-Tenant AI Agents covers what goes wrong when tenant boundaries are not enforced server-side, and agent tool calling auth production problems and patterns covers why a shared-credential model creates systemic risk for the whole team.
Where the Claude Agent SDK Stops and Agent Auth Begins
The SDK's tool primitive is create_sdk_mcp_server: define functions with @tool, bundle them, hand the server to the agent. Clean model, single-user defaults.
create_sdk_mcp_server(tools=[...])
Tools are defined once, statically, at startup
Every rep gets the same surface; there is no per-identity surface
list_scoped_tools returns what this rep's connected account authorizes
@tool handler async def handler(args)
The handler carries its own credential
One Lusha key serves the whole team; one quota, one credit pool, one blast radius
execute_tool resolves that rep's vaulted key server-side, per call
The input_schema you pass to @tool
Every parameter is the model's to fill
email, seeds, and exclude become model-controlled
Strip them from the schema; inject resolved values in the closure
You can enumerate names ahead of time
Names are minted per rep at runtime, and long ones exceed the API's 64-character tool-name limit
Derive the allowlist from what list_scoped_tools returned, then assert the length
The rest of this build is one conversion: take the SDK's static, ambient-credential tool surface and make it per-run, identity-bound, and budget-bound. The model supplies judgment. The runtime supplies identity, tenancy, and spend.
Prerequisites
- Python 3.10 or newer, and pip install claude-agent-sdk scalekit-sdk-python protobuf python-dotenv. On clean virtualenvs install protobuf explicitly; some images do not pull it in.
- An ANTHROPIC_API_KEY. The Python SDK drives the Claude Code runtime underneath; install the CLI if the SDK prompts for it.
- A Scalekit account with a Lusha MCP connection created under AgentKit > Connections. Lusha uses API key authentication, so there is no redirect URI and no consent screen.
- A Lusha API key per rep, taken from API & connectors > Manage API Keys in Lusha. Set a monthly credit limit on each key before anyone connects; it is the only ceiling Lusha itself enforces.
- The connection_name in your code must match the connection name in the Scalekit dashboard character for character, including any suffix added at creation. A mismatch does not fail at startup; it surfaces as a not-found on the first execute_tool.
- Verified against scalekit-sdk-python 2.17.0 and claude-agent-sdk 0.2.148. Two signatures moved recently enough to be worth checking with inspect.signature before you copy anything.
# .env
SCALEKIT_ENVIRONMENT_URL=https://your-env.scalekit.cloud
SCALEKIT_CLIENT_ID=skc_...
SCALEKIT_CLIENT_SECRET=sks_...
ANTHROPIC_API_KEY=sk-ant-...
ANTHROPIC_MODEL=claude-sonnet-4-6
# must match Scalekit Dashboard > AgentKit > Connections exactly
LUSHA_CONNECTION=lushamcp
# resolved from your authenticated session in production; env vars here for one run
REP_IDENTIFIER=user_123
REP_LUSHA_EMAIL=priya@example.com
RUN_CREDIT_CEILING=400
Step 1: Vault the Rep's Key, Then Resolve the Run
One run is bound to five facts, and none of them come from the model: the Scalekit identifier that selects the vaulted key, the Lusha login email that owns any table written, the seed set, the exclusion set, and the credit ceiling.
Two SDK details decide whether this code works at all. For an API-key connector the credential goes in authorization_details under a static_auth key; the parameter list absorbs unknown keywords into **kwargs, so a misnamed argument creates a connected account with no key and fails later, inside a tool call. And get_connected_account_details returns status as a string, so comparing it against the imported ACTIVE protobuf enum (which is the integer 1) never matches.
# lusha_identity.py
import os
from dataclasses import dataclass
from dotenv import load_dotenv
from scalekit.client import ScalekitClient
from scalekit.common.exceptions import ScalekitNotFoundException
load_dotenv()
# Must match the connection name in the Scalekit dashboard, exactly.
LUSHA_CONNECTION = os.environ["LUSHA_CONNECTION"]
scalekit = ScalekitClient(
env_url=os.environ["SCALEKIT_ENVIRONMENT_URL"],
client_id=os.environ["SCALEKIT_CLIENT_ID"],
client_secret=os.environ["SCALEKIT_CLIENT_SECRET"],
)
actions = scalekit.actions
@dataclass(frozen=True)
class ExpansionRun:
"""Everything one run is bound to. Not one field is model-supplied."""
identifier: str # your user id; the key for the Scalekit connected account
owner_email: str # Lusha login email; the Tables ownership plane
seeds: dict # {"domains": [...]} from your own closed-won accounts
exclude: dict # {"domains": [...]} customers and open opportunities
credit_ceiling: int # hard cap on Lusha credits this run may spend
lookalike_page_limit: int = 50 # per-page `limit`; lookalikes bill per result
max_lookalike_pages: int = 3 # dedupeSessionId pages allowed per run
max_widen_pages: int = 2 # prospecting pages allowed per run
shortlist_cap: int = 25 # enrich caps ids at 25; keep scoring in step
def vault_rep_key(identifier: str, lusha_api_key: str) -> str:
"""Store this rep's Lusha key in the vault. Called from your integrations page.
Lusha is an API-key connector, so there is no consent redirect: the rep pastes
a key, you upsert it, and Scalekit holds it. `static_auth` is the shape the SDK
converts into the connector's credential struct; on scalekit-sdk-python 2.17.0
the parameter is `authorization_details`, and any other keyword name is
silently swallowed by **kwargs.
"""
response = actions.upsert_connected_account(
connection_name=LUSHA_CONNECTION,
identifier=identifier, # never accepted from client input
authorization_details={"static_auth": {"username": lusha_api_key}},
)
return response.connected_account.status
def lusha_preflight(identifier: str) -> tuple[bool, str]:
"""Deterministic gate, run before a single model token is spent."""
try:
# `_details` returns metadata only. The sibling `get_connected_account`
# returns the credential itself, which nothing in an agent runtime needs.
details = actions.get_connected_account_details(
connection_name=LUSHA_CONNECTION,
identifier=identifier,
)
except ScalekitNotFoundException:
return False, "no_connected_account"
# ConnectorStatus arrives as a string: ACTIVE, EXPIRED, PENDING_AUTH,
# PENDING_VERIFICATION, DISCONNECTED.
status = details.connected_account.status
if status != "ACTIVE":
return False, f"connected_account_status={status}"
# One account_usage call per run. Lusha rate limits this endpoint to 5
# requests per minute, and it is the only free look at the credit pool.
usage = actions.execute_tool(
tool_name="lushamcp_account_usage",
identifier=identifier,
tool_input={"reason_for_invocation": "Pre-run credit and rate-limit check"},
connection_name=LUSHA_CONNECTION,
)
return True, str(usage.data)
def build_run(
identifier: str,
owner_email: str,
seed_domains: list[str],
customer_domains: list[str],
credit_ceiling: int,
) -> ExpansionRun:
"""Validate the tenant's own data against Lusha's ceilings, before the model runs.
Seeds and exclusions are tenant records, not model output. Lusha rejects a
seed set outside 5-100 identifiers, and caps `exclude` at 500. Above 500
customers, exclude the remainder against your own store after the call.
"""
seeds = sorted({d.strip().lower() for d in seed_domains if d and d.strip()})
if not 5 <= len(seeds) <= 100:
raise ValueError(f"Lusha lookalikes need 5-100 seed identifiers; got {len(seeds)}")
exclude = sorted({d.strip().lower() for d in customer_domains if d and d.strip()})
if len(exclude) > 500:
raise ValueError(f"exclude caps at 500 identifiers; got {len(exclude)}")
return ExpansionRun(
identifier=identifier,
owner_email=owner_email,
seeds={"domains": seeds},
exclude={"domains": exclude},
credit_ceiling=credit_ceiling,
)
The tenant boundary rests entirely on identifier and owner_email being trustworthy. Resolve both server-side; never accept either from client input. Access Control for Multi-Tenant AI Agents covers what goes wrong when that rule slips.
Step 2: Cut 42 Tools to Seven, and Check the Name Budget
actions.tools.list_scoped_tools returns the tools this rep's connected account authorizes, already in Anthropic's native format. That is the identity gate. The role gate is yours, and on this connector it is a correctness gate before it is an economy one.
- lushamcp_recommendations_companies, ..._contacts, and their two filter tools are OAuth-only; the connector schema states an API-key session receives a 403. An expansion agent that reaches for ICP recommendations burns turns on a wall.
- Five tools carry the STALE marker. lushamcp_prospecting_company_search in particular still describes 18 usable filter parameters, which makes it the most attractive wrong choice on the surface.
- Twelve Tables tools include lushamcp_table_delete, which permanently removes a table and its rows, and lushamcp_table_get_entities, which bills per row returned.
- lushamcp_companies_search defaults enrich to true, spending reveal credits inside what reads like a lookup.
At roughly 200 tokens per schema, 42 tools is about 8,400 tokens of surface before the agent does any work; seven is about 1,400. The honest caveat: tool search is on by default in the Agent SDK and defers full schemas until Claude loads one, which softens the token argument. It does not touch the other two. Forty-two similarly prefixed names is a decision space the model was not designed to handle at that scale, and a deferred lushamcp_table_delete is still a callable lushamcp_table_delete. Surface reduction is the lever. Model upgrades help; they are not the lever.
One registration-time failure is worth an assertion rather than a comment. MCP tools reach the model as mcp__{server}__{tool}, and the API caps tool names at 64 characters; the Claude Code issue tracker has repeated reports of long names producing a 400 and, in some cases, bricking the conversation until a new session starts. lushamcp_prospecting_company_search_by_text is 43 characters on its own, which leaves exactly 14 for the server name.
# lusha_surface.py
from google.protobuf.json_format import MessageToDict
from lusha_identity import LUSHA_CONNECTION, actions
SERVER_NAME = "lusha_expand" # 12 chars; see MAX_TOOL_NAME_CHARS below
MAX_TOOL_NAME_CHARS = 64 # Anthropic API limit on tool_name
# Seven of the connector's 42 tools. Every omission is deliberate: the
# recommendations_* tools 403 on an API key, table_get_entities bills per row
# read, table_delete is destructive, and companies_search enriches by default.
EXPANSION_TOOLS = {
"lushamcp_account_usage",
"lushamcp_lookalike_companies",
"lushamcp_prospecting_company_search_by_text",
"lushamcp_prospecting_company_enrich",
"lushamcp_signal_score_companies",
"lushamcp_table_create",
"lushamcp_table_add_entities",
}
# Identity and tenant data. These are claims, not parameters, so the model
# never sees them in a schema and cannot populate them.
SERVER_INJECTED = {"email", "seeds", "exclude"}
def _harden_schema(schema: dict) -> dict:
"""Remove server-injected fields; promote the audit field to required.
`reason_for_invocation` is optional upstream. Making it required here is what
turns the spend gate in step 4 into a rule with an artifact behind it.
"""
properties = {
k: v for k, v in schema.get("properties", {}).items()
if k not in SERVER_INJECTED
}
required = [r for r in schema.get("required", []) if r not in SERVER_INJECTED]
if "reason_for_invocation" in properties and "reason_for_invocation" not in required:
required.append("reason_for_invocation")
return {**schema, "properties": properties, "required": required}
def _assert_name_fits(tool_name: str) -> str:
"""Fail at registration, not mid-run with a 400 the model cannot recover from."""
qualified = f"mcp__{SERVER_NAME}__{tool_name}"
if len(qualified) > MAX_TOOL_NAME_CHARS:
raise ValueError(
f"{qualified} is {len(qualified)} chars against a {MAX_TOOL_NAME_CHARS} "
f"limit. Shorten SERVER_NAME by {len(qualified) - MAX_TOOL_NAME_CHARS}."
)
return qualified
def expansion_surface(identifier: str) -> list[dict]:
"""Identity gate (Scalekit), then role gate (this function), in that order."""
scoped_response, _ = actions.tools.list_scoped_tools(
identifier=identifier,
filter={"connection_names": [LUSHA_CONNECTION]},
page_size=100, # fetch past the default page so no tool is missed
)
surface, seen = [], set()
for item in scoped_response.tools:
definition = MessageToDict(item.tool).get("definition", {})
name = definition.get("name")
if name not in EXPANSION_TOOLS:
continue # role gate: 42 authorized tools down to 7
_assert_name_fits(name)
seen.add(name)
surface.append({
"name": name,
"description": definition.get("description", ""),
"input_schema": _harden_schema(definition.get("input_schema", {})),
})
# The filter also accepts `tool_names`, which would drop missing tools
# silently. Asserting the delta here is how upstream renames surface as a
# deploy failure instead of a run that quietly skips the widen step.
missing = EXPANSION_TOOLS - seen
if missing:
raise RuntimeError(f"Not authorized, or renamed upstream: {sorted(missing)}")
return surface
A third layer applies at execution and is invisible in this file: because every call runs against that rep's own key, Lusha's plan entitlements and per-key credit cap still apply. What the rep cannot do in Lusha, the agent cannot do.
Recommended reading: How to Implement Least Privilege for AI Agent Tool Calls.
Step 3: Bind the Identity and the Tenant Data the Model Never Sees
The @tool handler signature is async def handler(args). It receives the model's arguments and nothing else, so the rep and their seed set have to be captured in a closure when the tool is built, per run. Build the server once at import time with a module global and every rep writes into one person's workspace off one person's seed list; that is the opening failure, reintroduced one layer up.
This handler also does the accounting, because it is the only place the response body actually lands. billing.creditsCharged is present on V3 responses; when it is absent, the pre-call estimate stands in rather than counting zero.
# lusha_tools.py
import asyncio
import json
from claude_agent_sdk import ToolAnnotations, tool
from lusha_budget import CreditLedger, estimate_credits
from lusha_identity import LUSHA_CONNECTION, ExpansionRun, actions
TABLE_TOOL_PREFIX = "lushamcp_table_"
def make_lusha_tool(tool_def: dict, run: ExpansionRun, ledger: CreditLedger):
"""Wrap one Scalekit tool as an in-process SDK tool bound to one rep.
`execute_tool` resolves that rep's vaulted Lusha key server-side; no key
enters this process, this handler, or the model context.
"""
name = tool_def["name"]
@tool(
name,
tool_def["description"],
tool_def["input_schema"],
# Every billable call is marked non-read-only so nothing about the
# surface invites parallel invocation. Annotations are hints and the SDK
# docs say clients should not rely on them for security decisions, which
# is why the ledger below takes a lock regardless.
annotations=ToolAnnotations(readOnlyHint=(name == "lushamcp_account_usage")),
)
async def _handler(args: dict) -> dict:
payload = dict(args)
# Ownership plane. Set here and nowhere else. Lusha resolves this email
# to a user on the account behind the key, so a model-chosen value is a
# cross-user write that returns 200.
if name.startswith(TABLE_TOOL_PREFIX):
payload["email"] = run.owner_email
# Tenant data plane. `exclude` is applied on every request and combined
# with server-side dedupe, so injecting it per call keeps existing
# customers out even when the model starts a fresh dedupe session.
if name == "lushamcp_lookalike_companies":
payload["seeds"] = run.seeds
payload["exclude"] = run.exclude
estimated = estimate_credits(name, payload, run, ledger)
try:
# execute_tool is blocking; keep the agent loop responsive.
result = await asyncio.to_thread(
actions.execute_tool,
tool_name=name,
identifier=run.identifier, # the acting rep, per run
tool_input=payload,
connection_name=LUSHA_CONNECTION,
)
except Exception as exc:
async with ledger.lock:
ledger.reserved = max(0, ledger.reserved - estimated)
ledger.trail.append({"tool": name, "ok": False, "error": str(exc)})
# A revoked key, a 402, or a 451 should degrade this step, not kill
# the run. Catching it lets you compose the message Claude reads.
return {
"content": [{"type": "text", "text": f"{name} failed: {exc}"}],
"is_error": True,
}
body = result.data or {}
charged = int((body.get("billing") or {}).get("creditsCharged") or estimated)
async with ledger.lock:
ledger.reserved = max(0, ledger.reserved - estimated)
ledger.spent += charged
# Harvest canReveal so step 4 can refuse a reveal the run has never
# seen priced. A field priced at 0 is priced; an absent field is not.
for row in body.get("data") or []:
company_id = str(row.get("id") or "")
if company_id:
ledger.revealable[company_id] = {
f["field"]: int(f.get("credits") or 0)
for f in row.get("canReveal") or []
}
ledger.trail.append({
"identifier": run.identifier,
"tool": name,
"execution_id": result.execution_id,
"credits": charged,
"reason": args.get("reason_for_invocation"),
})
return {"content": [{"type": "text", "text": json.dumps(body, default=str)}]}
return _handler
Keys live in Scalekit's vault and are injected server-side inside execute_tool; see token vaults for AI agent workflows for why keeping credentials out of the model context matters more, not less, when the credential is a static key with no expiry.
Step 4: Make the Credit Budget a Permission, Not a Prompt Instruction
On a credit-metered connector, authorization is not only "may this agent call this tool." It is "may this agent spend this much, of whose budget." A system prompt cannot answer the second question, because the model has a plausible reason to page one more time on every turn.
Two SDK facts decide which control to use. allowed_tools auto-approves, so can_use_tool is never invoked for a tool you allowlisted; the Agent SDK hooks reference is explicit that a PreToolUse hook is what gates every call. That hook can also return updatedInput, which means an over-large limit gets clamped rather than refused, and the run keeps moving.
# lusha_budget.py
import asyncio
from dataclasses import dataclass, field
# Tools that move credits. This classification is your policy; the connector
# does not declare it, so keep it beside the code that enforces it.
BILLABLE = {
"lushamcp_lookalike_companies",
"lushamcp_prospecting_company_search_by_text",
"lushamcp_prospecting_company_enrich",
"lushamcp_signal_score_companies",
"lushamcp_table_add_entities",
}
COMPANY_REVEAL_BASE = 1 # per result on company enrich
WIDEN_PAGE_WORST_CASE = 50 # no page-size dial exists; assume a full page
SIGNAL_WORST_CASE_PER_COMPANY = 4 # signals bill per matched signal; match count
# is not knowable pre-call, so this is a
# policy number, not a Lusha-published price
@dataclass
class CreditLedger:
"""Per-run spend state. Constructed with the tools, never shared across reps."""
ceiling: int
spent: int = 0
reserved: int = 0
revealable: dict = field(default_factory=dict) # company_id -> {field: credits}
calls: dict = field(default_factory=dict) # tool_name -> count
trail: list = field(default_factory=list)
lock: asyncio.Lock = field(default_factory=asyncio.Lock)
@property
def headroom(self) -> int:
return self.ceiling - self.spent - self.reserved
def price_of(self, company_id, fields) -> int:
prices = self.revealable.get(str(company_id), {})
return sum(prices.get(f, 0) for f in fields)
def unpriced(self, company_ids, fields) -> list[str]:
"""Fields never seen in a canReveal preview for one of these ids."""
out = set()
for company_id in company_ids:
seen = self.revealable.get(str(company_id), {})
out |= {f for f in fields if f not in seen}
return sorted(out)
def estimate_credits(tool_name: str, args: dict, run, ledger: CreditLedger) -> int:
"""Worst-case cost before the call. Reconciled against billing.creditsCharged after."""
if tool_name == "lushamcp_lookalike_companies":
return max(1, min(int(args.get("limit") or 25), run.lookalike_page_limit))
if tool_name == "lushamcp_prospecting_company_search_by_text":
return WIDEN_PAGE_WORST_CASE
if tool_name == "lushamcp_prospecting_company_enrich":
ids = args.get("ids") or []
fields = args.get("reveal") or []
return sum(COMPANY_REVEAL_BASE + ledger.price_of(i, fields) for i in ids)
if tool_name == "lushamcp_signal_score_companies":
return len(args.get("companies") or []) * SIGNAL_WORST_CASE_PER_COMPANY
if tool_name == "lushamcp_table_add_entities":
return len(args.get("entity_ids") or []) # companies bill per newly-added
return 0
def _deny(reason: str) -> dict:
"""permissionDecisionReason is read by Claude, so it doubles as the correction."""
return {"hookSpecificOutput": {
"hookEventName": "PreToolUse",
"permissionDecision": "deny",
"permissionDecisionReason": reason,
}}
def _clamp(updated: dict, note: str) -> dict:
return {"hookSpecificOutput": {
"hookEventName": "PreToolUse",
"permissionDecision": "allow",
"permissionDecisionReason": note,
"updatedInput": updated,
}}
def make_spend_gate(run, ledger: CreditLedger):
"""Fires on every tool call, because every tool on this connector has a price."""
async def gate(input_data, tool_use_id, context):
# Hooks receive the qualified mcp__server__tool name.
name = input_data.get("tool_name", "").rsplit("__", 1)[-1]
args = dict(input_data.get("tool_input") or {})
async with ledger.lock:
ledger.calls.setdefault(name, 0)
if name in BILLABLE and not (args.get("reason_for_invocation") or "").strip():
return _deny(
"Every billable Lusha call must carry reason_for_invocation. "
"Write it for a human auditor reading the log next quarter."
)
if name == "lushamcp_account_usage" and ledger.calls[name] >= 1:
return _deny(
"account_usage was already read for this run; Lusha rate limits "
"it to 5 requests per minute. Use the headroom already reported."
)
if name == "lushamcp_lookalike_companies":
if ledger.calls[name] >= run.max_lookalike_pages:
return _deny(
f"{run.max_lookalike_pages} lookalike pages already drawn. "
"Each page bills per result returned. Rank what you have."
)
if int(args.get("limit") or 25) > run.lookalike_page_limit:
args["limit"] = run.lookalike_page_limit
ledger.calls[name] += 1
ledger.reserved += estimate_credits(name, args, run, ledger)
return _clamp(args, f"limit clamped to {run.lookalike_page_limit}.")
if name == "lushamcp_prospecting_company_search_by_text":
if ledger.calls[name] >= run.max_widen_pages:
return _deny(
f"The widen step is capped at {run.max_widen_pages} pages; "
"this tool has no page-size parameter to reduce instead."
)
if name == "lushamcp_prospecting_company_enrich":
ids = args.get("ids") or []
if len(ids) > 25:
return _deny("prospecting_company_enrich caps ids at 25 per call.")
unpriced = ledger.unpriced(ids, args.get("reveal") or [])
if unpriced:
# Not an assertion about what V3 accepts; an assertion that the
# agent only asks to spend on fields it has seen priced.
return _deny(
f"Fields {unpriced} were not priced in a canReveal preview "
"for these ids. Omit reveal to take the free intent field."
)
if name == "lushamcp_signal_score_companies":
if len(args.get("companies") or []) > run.shortlist_cap:
return _deny(
f"Score at most {run.shortlist_cap} companies; signals bill "
"per matched signal per company."
)
if name == "lushamcp_table_add_entities":
if len(args.get("entity_ids") or []) > run.shortlist_cap:
return _deny(f"Write at most {run.shortlist_cap} companies.")
cost = estimate_credits(name, args, run, ledger)
if cost > ledger.headroom:
return _deny(
f"This call could cost {cost} Lusha credits and {ledger.headroom} "
"remain in this run's budget. Summarise the ranked accounts and stop."
)
ledger.calls[name] += 1
ledger.reserved += cost
return {} # no opinion; fall through to the allowlist
return gate
def make_budget_reporter(ledger: CreditLedger):
"""Tell the model what is left, so it stops planning spend it cannot make."""
async def report(input_data, tool_use_id, context):
return {"hookSpecificOutput": {
"hookEventName": "PostToolUse",
"additionalContext": (
f"Lusha credit headroom remaining for this run: {ledger.headroom}."
),
}}
return report
The tradeoff is real and worth naming. Capping lookalike pages at three trades recall for containment: an expansion agent that could have surfaced a fourth page of candidates will not. Raise the ceiling for a deliberate quarterly list build and lower it for an interactive run; what you should not do is let the model choose the boundary. Where a human genuinely has to authorise the spend, human-in-the-loop tool calling is the pattern to reach for.
Step 5: Contain Everything Else and Run the Loop
tools=[] is the line people skip. allowed_tools changes permission, not availability; without tools=[] the entire Claude Code built-in set, Bash included, stays in context on a host holding B2B contact data. permission_mode="dontAsk" matters for the same reason: on an unattended run, a prompt is a hang.
Note the two budgets. max_budget_usd stops model spend and is compared against the client-side cost estimate; the ledger stops Lusha credit spend. Neither substitutes for the other.
# lusha_agent.py
import asyncio
import os
from claude_agent_sdk import (
AssistantMessage,
ClaudeAgentOptions,
ClaudeSDKClient,
HookMatcher,
ResultMessage,
TextBlock,
create_sdk_mcp_server,
)
from lusha_budget import CreditLedger, make_budget_reporter, make_spend_gate
from lusha_identity import build_run, lusha_preflight
from lusha_surface import SERVER_NAME, expansion_surface
from lusha_tools import make_lusha_tool
EXPANSION_SYSTEM_PROMPT = """You build a ranked account expansion list in Lusha for one
sales rep. Seeds and exclusions are already set; you cannot change them.
Work in this order and stop when the list is written.
1. Read the credit headroom reported to you. Never call account_usage twice.
2. Call lushamcp_lookalike_companies to draw candidates from the seed set. Start with
the default limit. Draw a second page only if the first returns fewer usable
candidates than asked for, and pass back the dedupeSessionId from the previous
response so you do not pay for duplicates.
3. Only if lookalikes are thin, call lushamcp_prospecting_company_search_by_text once
with a description of the target audience in plain language. This tool has no
filter parameters and no page size, so make the description precise.
4. Shortlist at most 25 candidates on the preview fields alone. Then call
lushamcp_prospecting_company_enrich on those ids. Omit `reveal` unless you have
read a canReveal entry for that id and the field is worth its credit cost.
5. Call lushamcp_signal_score_companies once on the shortlist to rank it. Order the
final list by signal strength first and firmographic fit second.
6. Call lushamcp_table_create for a companies table named for this expansion run,
then lushamcp_table_add_entities once with the ranked ids.
Every call takes reason_for_invocation, and it is required. State which step you are
on and why these specific accounts. Never claim an account qualifies on a field you
have not read back from a response. If a tool is denied, do not call it again with
different arguments; report what you have and stop."""
def build_options(surface, run, ledger):
server = create_sdk_mcp_server(
name=SERVER_NAME,
version="1.0.0",
tools=[make_lusha_tool(t, run, ledger) for t in surface],
)
# Derived from what list_scoped_tools returned, never typed by hand.
allowed = [f"mcp__{SERVER_NAME}__{t['name']}" for t in surface]
return ClaudeAgentOptions(
mcp_servers={SERVER_NAME: server},
strict_mcp_config=True, # ignore .mcp.json, user settings, any other server
allowed_tools=allowed, # auto-approve exactly this surface
tools=[], # remove every built-in: no Bash, Write, WebFetch
setting_sources=[], # no filesystem settings (needs Python SDK > 0.1.59)
permission_mode="dontAsk", # unattended: deny, do not prompt and hang
hooks={
# No matcher: the gate sees every call, because every tool has a price.
"PreToolUse": [HookMatcher(hooks=[make_spend_gate(run, ledger)])],
"PostToolUse": [HookMatcher(hooks=[make_budget_reporter(ledger)])],
},
max_turns=30,
max_budget_usd=2.00, # model spend, not Lusha credits
model=os.environ.get("ANTHROPIC_MODEL", "claude-sonnet-4-6"),
system_prompt=EXPANSION_SYSTEM_PROMPT,
)
async def run_expansion(
identifier: str,
owner_email: str,
seed_domains: list[str],
customer_domains: list[str],
credit_ceiling: int,
) -> dict:
ok, detail = lusha_preflight(identifier)
if not ok:
# EXPIRED, DISCONNECTED, or a missing account is a hard stop that notifies
# the rep, not something a scheduled run should retry.
raise PermissionError(f"Lusha is not callable for {identifier}: {detail}")
run = build_run(identifier, owner_email, seed_domains, customer_domains, credit_ceiling)
ledger = CreditLedger(ceiling=run.credit_ceiling)
surface = expansion_surface(identifier)
options = build_options(surface, run, ledger)
print(f"{len(surface)} tools registered for {run.identifier}; "
f"{run.credit_ceiling} credit ceiling")
prompt = (
f"Build a ranked expansion list from the {len(run.seeds['domains'])} seed "
f"accounts already configured, excluding the "
f"{len(run.exclude['domains'])} accounts we already work with."
)
answer = None
async with ClaudeSDKClient(options=options) as client:
await client.query(prompt)
async for message in client.receive_response():
if isinstance(message, AssistantMessage):
for block in message.content:
if isinstance(block, TextBlock):
print(block.text)
elif isinstance(message, ResultMessage):
# error_max_budget_usd means the model budget stopped the run;
# the credit ledger is a separate ceiling with its own outcome.
answer = {"subtype": message.subtype, "result": message.result}
return {
"answer": answer,
"credits_spent": ledger.spent,
"credits_remaining": ledger.headroom,
"calls": ledger.calls,
"trail": ledger.trail,
}
if __name__ == "__main__":
# In production every argument here comes from your own records, resolved
# server-side after the request is authenticated.
out = asyncio.run(run_expansion(
identifier=os.environ["REP_IDENTIFIER"],
owner_email=os.environ["REP_LUSHA_EMAIL"],
seed_domains=[
"acme.com", "globex.com", "initech.com", "umbrella.com",
"hooli.com", "piedpiper.com", "stark.com", "wayne.com",
],
customer_domains=["acme.com", "globex.com"],
credit_ceiling=int(os.environ.get("RUN_CREDIT_CEILING", "400")),
))
print(f"credits: {out['credits_spent']} spent, {out['credits_remaining']} left")
print(f"calls: {out['calls']}")
Change REP_IDENTIFIER and REP_LUSHA_EMAIL and everything underneath changes with it: a different vaulted key, a different quota, a different credit pool, a different owner on the table that gets written. No code changes. Adding a CRM read to the seed resolution adds a connection name and a few tool names; it adds no auth code, because the vault, the scoping, and the per-action audit trail are identical for every connector.
What Breaks Without This
One Lusha key in .env, static @tool handlers
Works for whoever owns the key
One quota and one credit cap for the whole team; a 402 in one rep's run was caused by another rep's Monday
email left in the input_schema
Model fills it correctly under test
The expansion table lands in whichever rep's address was in context, and Lusha returns 200
seeds and exclude left to the model
Accounts you already own reappear as "expansion," and a fresh dedupe session silently drops the exclusion
dedupeSessionId paging with no page cap
Pages until the monthly cap is gone; each page bills per result returned
reveal chosen freely on enrich
400 on fields absent from canReveal, and 5 credits per phone on the ones that do resolve
allowed_tools set, no tools=[]
Correct tools used in testing
Bash, Write, and WebFetch stay in context on a host holding contact data
Full 42-tool surface registered
recommendations_* returns 403, the STALE prospecting search looks like the best option, and table_delete is one confident wrong turn away
Server name chosen for readability
mcp__lusha_expansion__lushamcp_prospecting_company_search_by_text is 65 characters and the API rejects it
FAQs
Why not just tell the model in the system prompt to stop at three lookalike pages?
Because paging is the correct next action from the model's point of view on every turn, and it has a dedupeSessionId in hand that makes the next page cheap-looking and non-duplicative. You would be asking it to prefer your instruction over an available, sensible, on-task move, on every call, forever. The PreToolUse deny is one branch and has no failure mode. Prompts express intent; hooks express authority.
Should I use can_use_tool or a PreToolUse hook for the spend gate?
The hook. can_use_tool fires only when the permission flow falls through to a prompt, and every tool in allowed_tools is auto-approved before it gets there. Dropping the tools from the allowlist to force the callback gives you a prompt on every call, which defeats an unattended run.
Can one ledger and one server serve the whole sales team?
No. The rep is closed over in the tool handlers, the ledger holds that run's canReveal prices and page counts, and the owner email is baked into the table writes. Reuse the process; mint the surface, the options, and the ledger per run, per rep. How Tool Calling Auth Changes When You Move from Single-Tenant to Multi-Tenant covers the general shape of that boundary.
A rep hits their per-key monthly cap mid-run. What does the agent see?
A 402. Lusha's help centre notes the common confusion here: the rejection reads as a credit problem while the account dashboard still shows credits, because the cap is set per key rather than per account. Treat 402 as a hard stop that notifies the rep, and keep it distinct from 429 (rate or daily quota, worth a backoff) and 451 (blocked under GDPR, which is a legitimate empty result rather than a retry).
Why is lushamcp_table_get_entities not in the surface? The agent could verify its own write.
Because reading rows bills per row returned. An agent that re-reads a 25-row table to confirm the write pays for the reassurance, every run. Verify the write from execute_tool's response and the tableWrite confirmation in it; verify it independently in your own store, not by paying Lusha to read back what you just sent.
Does the Lusha key ever reach the model context?
No. It stays in Scalekit's vault and is resolved server-side inside execute_tool. The handler receives arguments and returns results; it never holds a credential. That matters more on an API-key connector than on OAuth, because a static key has no expiry to limit the damage: a key that leaks into a transcript is valid until an Admin rotates it, and rotating it disconnects every rep who depends on it.
How do I answer "which rep's agent spent 4,000 credits last Tuesday"?
Three fields joined on the connected account: the identifier the call ran as, the execution_id Scalekit returns on every execute_tool response, and the reason_for_invocation the model was required to write. The first two are facts; the third is model output, which is exactly what you need when a list looks correct and is not. Audit Trails for Agent Auth in B2B SaaS covers the event taxonomy, and Agent Tool Observability covers separating connector errors from infrastructure errors at the point they occur.
Should this run headless on a schedule?
The grant persists in the vault, so the agent can act as the rep without them online. What must not be headless is failure. Gate every scheduled run on lusha_preflight, and treat EXPIRED, DISCONNECTED, or a missing connected account as a stop that notifies the rep. Revoking an employee's AI agent access covers why the connected account, not the Lusha dashboard, is the revocation surface with the right granularity.
Next Steps to Start Building Your Lookalike Expansion Agent
- Create the connection under AgentKit > Connections, find Lusha MCP, and copy the exact connection name into LUSHA_CONNECTION. Set a monthly credit limit on each Lusha key in API & connectors > Manage API Keys before any rep connects.
- Install with pip install claude-agent-sdk scalekit-sdk-python protobuf python-dotenv, set ANTHROPIC_API_KEY, then run vault_rep_key for one rep and confirm lusha_preflight returns ACTIVE before writing agent logic.
- Call expansion_surface(identifier) and assert you get seven tools, not 42 and not zero. Zero means the connected account is not ACTIVE; a RuntimeError naming a missing tool means the connector renamed it upstream and your role gate caught it.
- Ship the run with tools=[], strict_mcp_config=True, a derived allowed_tools, and both hooks registered. Then open Lusha and confirm the expansion table is owned by the rep who ran the agent, and that account_usage shows the credits the ledger reported.
- Verify the same attribution independently in the Scalekit agent audit logs: who authorized, which agent ran, which tool, what came back, what it cost.
- Add the buying committee at the top-ranked accounts with lushamcp_buying_group_search, which bills per contact returned; put contactsLimit and the personas filter behind the same spend gate before you register it.
- Push the finished list somewhere a rep works.
Use the outbound prospecting agent walkthrough for the sequencer path and the CRM AI Agent template for writing qualified accounts back.
Building something on Lusha and want a second pair of eyes on the credential model? Join the Scalekit Slack community, or talk to an engineer.
.png)
