TL;DR
- The Claude Agent SDK gives you an autonomous reasoning loop, but its tool model (@tool + create_sdk_mcp_server) assumes static, in-process tools with ambient credentials; that is a single-user assumption, and wiring a per-rep deal-risk agent that way ships a cross-tenant data path that no prompt can close.
- Correct multi-user behavior is set by how you mint the tool surface and bind identity per run, not by the model: actions.tools.list_scoped_tools defines the per-rep surface, execute_tool binds every call to that rep's connected account, and no Gong, Attio, or Slack token ever enters the agent process or the model context.
- Because a reasoning loop can select any registered tool, blast radius is a first-class control: strip built-ins with tools=[], register only the deal-risk read tools plus attio_create_note, attio_create_task, and slack_send_message, and derive allowed_tools from the discovered tool names rather than a hardcoded list.
- allowed_tools is a permission allowlist, not an availability filter; it does not remove built-in tools from Claude's toolset, so availability control and identity-bound execution are two separate problems you solve explicitly.
- Scalekit AgentKit supplies the two hinges (list_scoped_tools and execute_tool) that turn the SDK's single-user tool model into a per-rep, least-privilege one, with a per-action audit trail tying every Gong, Attio, and Slack call to the rep who authorized it.
- The full code is on GitHub. Clone the repo, configure your Attio, Gong, and Slack connections, and have it running in under 30 minutes.
A rep pastes one Attio token into .env, wires it into a Claude Agent SDK tool, and the deal-risk agent works in the demo: it reads the pipeline, scores every open deal, writes a risk note back to the slipping ones, and posts a ranked brief to Slack. Then the second rep runs it. The agent reads the first rep's pipeline, scores deals that belong to someone else's book, and writes a "high-risk, competitor mentioned twice" note into the wrong workspace under the wrong identity. Nothing throws. The run is green. The audit log, if there is one, shows a single service account touching every deal in the company.
That failure is not a bug in the reasoning. It is a property of how the agent was wired. The Claude Agent SDK hands you an autonomous loop for free; its tool-registration model assumes tools that are statically defined, in-process, and backed by ambient credentials. That is a single-user assumption. A deal-risk agent is inherently multi-rep and multi-workspace, and the gap between those two facts is exactly where the cross-tenant data path opens.
What This Agent Is, and Why It Is a Reasoning Loop
Scalekit's earlier deal-intelligence build was a deterministic pipeline: a scheduler ran the same fixed steps every morning (fetch calls, fetch transcripts, match deals, score, post) and exited. Every branch was hard-coded. This build is the other kind of agent. The Claude Agent SDK runs the same loop that powers Claude Code, so Claude decides which tool to call next and in what order, given the deal-risk objective and the tools it can see.
That autonomy is the point, and it is also why auth stops being a setup step and becomes a load-bearing control.
In a deterministic pipeline, the sequence of tool calls is fixed by your code, so an over-broad token is contained by the fact that you never call the dangerous tool. In a reasoning loop, Claude can call any tool that is registered and permitted. If attio_delete_deal is in the surface, one confident wrong turn deletes a deal. If the surface is a shared workspace token, one wrong turn crosses a tenant boundary. The scoring logic (weight sentiment, days-to-close, engagement, and objections into a 0.0 to 1.0 risk score) lives in the system prompt. The correctness that keeps a rep's agent inside a rep's data lives entirely in the tool surface and the identity behind each call.
The agent's job, per run, for one rep:
- Pull that rep's recent Gong calls and transcripts.
- Extract sentiment, engagement, competitor mentions, and objections.
- Cross-reference each call against that rep's Attio deals for stage, value, and close date.
- Score deal risk, write a concise risk note to the matched Attio deal, open a follow-up task on the high-risk ones, and post a ranked brief to Slack.
Every one of those reads and writes must execute as the rep, scoped to what the rep can see in Gong, Attio, and Slack. What the rep cannot do, the agent must not be able to do.
Where the Claude Agent SDK Stops and Agent Auth Begins
The SDK's tool primitive is create_sdk_mcp_server: you define functions with the @tool decorator, bundle them into an in-process MCP server, and hand that server to the agent. It is a clean model, and its defaults are single-user. Three assumptions matter here.
What breaks in a multi-rep agent
create_sdk_mcp_server(tools=[...])
Tools are defined once, statically, at startup
Every rep gets the same tool surface; there is no per-identity surface
list_scoped_tools returns only the tools the current rep's connected account authorizes
@tool handler async def handler(args)
The handler carries its own credential (env var, module global)
One ambient Gong/Attio/Slack token serves all reps; the agent acts as a shared account
execute_tool resolves the rep's vaulted token server-side, per call
You can enumerate tool names ahead of time
Names are minted per rep at runtime, so a static allowlist cannot cover them
Derive the allowlist from the names list_scoped_tools returned
The through-line for the rest of this build is a single conversion: take the SDK's static, ambient-credential tool surface and make it per-run and identity-bound. list_scoped_tools and execute_tool are the two points that conversion turns on. Everything below is that conversion, in order: resolve who, retrieve which tools, bind whose credentials, contain what it can do, then run.
Prerequisites
- Python 3.10 or newer (the Claude Agent SDK requires it).
- pip install claude-agent-sdk scalekit-sdk-python protobuf python-dotenv. On clean virtualenvs, install protobuf explicitly; scalekit-sdk-python lists it but some images do not pull it in.
- An ANTHROPIC_API_KEY. The Python SDK drives the Claude Code runtime under the hood; install the CLI if the SDK prompts for it.
- A Scalekit account (the free tier covers 1M MAU and 10K connected accounts) with three connections created under AgentKit > Connections: Attio, Gong, and Slack.
- For Attio: register your own OAuth app at build.attio.com (there is no shared managed app), set multi-workspace access to Yes, and select minimum scopes record_permission:read-write, note:read-write, task:read-write.
- For Gong: a workspace admin authorization with api:calls:read, api:calls:transcript:read, and api:users:read.
- For Slack: user-token scopes chat:write, channels:read, and channels:history so posts appear as the rep, not a bot.
One thing to get right before any code runs: the connection_name you pass in code must match the connection name in your Scalekit dashboard exactly, including any suffix added at creation (for example gong-abc12345). A mismatch routes to the wrong connection or returns a not-found error, and it is the single most common integration failure.
Environment variables:
# .env
# Use the exact variable name your dashboard shows under Developers > API Credentials.
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
# connection names must match the Scalekit dashboard exactly
ATTIO_CONNECTION=attio
GONG_CONNECTION=gong
SLACK_CONNECTION=slack
# Slack channel ID: right-click the channel > Copy Link > last path segment
SLACK_CHANNEL_ID=C0XXXXXXXXX
Step 1: Resolve Inbound Identity
Before the agent can see a single tool, it needs to know which rep it is acting for. That identity is the input to everything downstream: the tool surface is scoped to it, and every execution is bound to it.
get_or_create_connected_account is idempotent. The first call creates the per-rep account record in Scalekit; every later call returns the existing record with its current status. If the rep has not authorized a connector, get_authorization_link mints a consent URL on the spot, so a revoked or first-time connection is handled inline instead of failing mid-run.
import os
from scalekit.client import ScalekitClient
from dotenv import load_dotenv
load_dotenv()
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
# The three connections this agent role uses. The dict values must match the
# connection names configured in the Scalekit dashboard exactly.
CONNECTIONS = [
os.environ["ATTIO_CONNECTION"],
os.environ["GONG_CONNECTION"],
os.environ["SLACK_CONNECTION"],
]
def ensure_authorized(connection_name: str, identifier: str) -> bool:
"""Confirm the rep's connected account is ACTIVE for one connection.
Returns True if ready to use. If not authorized, prints a consent link and
returns False so the caller can stop before touching any provider API.
"""
account = actions.get_or_create_connected_account(
connection_name=connection_name,
identifier=identifier,
)
if account.connected_account.status == "ACTIVE":
print(f" ✓ {connection_name} ({identifier}) — ACTIVE")
return True
# Not authorized yet: mint a per-user consent URL scoped to this connection.
link = actions.get_authorization_link(
connection_name=connection_name,
identifier=identifier,
).link
print(f" ✗ {connection_name} not authorized for {identifier}. Authorize:\n {link}")
return False
The identifier is the rep's stable ID. In production you resolve it from the authenticated session (session cookie, verified JWT, or a database lookup) and never accept it from client input; the whole tenant boundary rests on that value being trustworthy. The same agent code serves every rep by changing only this string.
Step 2: Retrieve the Authorized Tool Surface
With the rep resolved, retrieve the tools their connected accounts authorize. actions.tools.list_scoped_tools returns exactly that: not a flat catalog of everything Attio, Gong, and Slack can do, but the tools this rep's connected accounts are permitted to call, already in Anthropic's native tool format (name, description, JSON Schema).
This is the accuracy and cost lever, not a convenience. A full connector catalog across three providers is well over a hundred tools; the Attio connector alone exposes around fifty. An LLM asked to pick the right tool from a surface that large selects worse and hallucinates parameters, and every tool in context burns tokens before the agent does any work. Scoping the surface to what one rep authorized shrinks the decision space to what is relevant for this rep, this connection, this task. The fix is surface reduction, not better prompting.
There are two independent gates, and it is worth being precise about which does what:
- Identity gate (Scalekit). list_scoped_tools returns only what the rep's connected accounts authorize. This scopes which tools appear.
- Role gate (your code). A deal-risk agent does not need every authorized tool; it needs a specific read set plus three writes. Intersect the scoped surface with an explicit role allowlist. This scopes which tools this agent is allowed to have, independent of identity.
A third layer applies at execution time and is not visible here: because every call runs against the rep's own token, the provider's own record-level access control still applies. list_scoped_tools controls the tool surface; the rep's token controls whose records each tool can touch.
from google.protobuf.json_format import MessageToDict
# The role allowlist: the only tools a deal-risk agent should ever hold.
# Read tools gather signal; the three write tools are the agent's entire
# authority to change state. Everything destructive (attio_delete_*,
# attio_update_deal stage mutations) is deliberately absent.
DEAL_RISK_TOOLS = {
# Gong: calls and transcripts for risk signal
"gong_calls_list",
"gong_calls_get",
"gong_calls_transcript_get",
# Attio: read deal context, write risk notes and follow-up tasks
"attio_search_records",
"attio_list_records",
"attio_list_deals",
"attio_get_deal",
"attio_list_notes",
"attio_create_note",
"attio_create_task",
# Slack: read prior context (optional) and post the brief
"slack_fetch_conversation_history",
"slack_send_message",
}
def discover_scoped_tools(identifier: str) -> list[dict]:
"""Return the intersection of (what the rep authorized) and (what a
deal-risk agent is allowed to hold), as Anthropic-native tool defs plus
the connection each tool routes to.
"""
scoped_response, _ = actions.tools.list_scoped_tools(
identifier=identifier,
filter={"connection_names": CONNECTIONS},
page_size=100,
)
tools = []
for item in scoped_response.tools:
definition = MessageToDict(item.tool).get("definition", {})
name = definition.get("name")
if name not in DEAL_RISK_TOOLS:
continue # role gate: drop anything outside the deal-risk set
tools.append(
{
"name": name,
"description": definition.get("description", ""),
"input_schema": definition.get("input_schema", {}),
# prefix maps the tool to its connection for execute_tool routing
"connection_name": name.split("_", 1)[0]
if name.split("_", 1)[0] in CONNECTIONS
else _connection_for(name),
}
)
return tools
def _connection_for(tool_name: str) -> str:
"""Map a tool name to its Scalekit connection by provider prefix."""
prefix = tool_name.split("_", 1)[0]
mapping = {
"gong": os.environ["GONG_CONNECTION"],
"attio": os.environ["ATTIO_CONNECTION"],
"slack": os.environ["SLACK_CONNECTION"],
}
return mapping[prefix]
The scoped surface is a function of the rep's identity; the role allowlist is a function of the agent's job. Neither is hardcoded to a specific rep, and the agent can never hold a tool that is outside both gates.
Step 3: Bind Identity Into the Surface
This is the hinge the whole build turns on. The SDK's @tool handler signature is async def handler(args); it receives the model's arguments and nothing else. It does not receive the acting rep. So the rep's identity has to be captured in a closure when the tool is built, per run, and every execute_tool call inside that handler carries it.
Get this wrong in the obvious way (build the server once at import time with a module-global identifier) and every rep's agent executes against that one identity: the shared-token failure from the opening, reintroduced one layer up. The server, and the identity baked into its handlers, must be minted per run.
Two more details matter at this layer, and both are auth-specific:
- actions.execute_tool is a synchronous, blocking call. Calling it directly inside an async handler blocks the event loop; offload it with asyncio.to_thread.
- When a rep revokes a connection or a token expires, the call fails. Return is_error: True with a readable message rather than letting the exception escape. An uncaught exception stops the agent loop; an is_error result lets Claude read the failure, skip that deal, and keep going. A revoked Gong connection should degrade the run, not crash it.
import asyncio
import json
from claude_agent_sdk import tool
def make_sdk_tool(tool_def: dict, identifier: str):
"""Wrap one Scalekit tool as an in-process SDK tool bound to `identifier`.
The rep's identity is closed over here, so this tool can only ever act as
this rep. execute_tool resolves the rep's vaulted token server-side; no
Gong/Attio/Slack credential enters this process or the model context.
"""
tool_name = tool_def["name"]
connection_name = tool_def["connection_name"]
# The Python @tool decorator accepts a full JSON Schema dict directly,
# so the Anthropic-native input_schema from Scalekit passes straight through.
@tool(tool_name, tool_def["description"], tool_def["input_schema"])
async def _handler(args: dict) -> dict:
try:
# execute_tool is blocking; keep the agent loop responsive.
result = await asyncio.to_thread(
actions.execute_tool,
tool_name=tool_name,
identifier=identifier, # <-- the acting rep, per run
tool_input=args,
connection_name=connection_name,
)
data = result.data or {}
return {"content": [{"type": "text", "text": json.dumps(data, default=str)}]}
except Exception as exc:
# Revoked/expired connection or provider error: fail closed for this
# tool, keep the run alive. Claude reads this and moves on.
return {
"content": [{"type": "text", "text": f"{tool_name} failed: {exc}"}],
"is_error": True,
}
return _handler
Step 4: Contain the Blast Radius
A reasoning loop chooses its own tool calls, so the containment question is not "what will it do" but "what is it able to do." Two facts about the SDK make this explicit work rather than a default.
First, allowed_tools is a permission allowlist, not an availability filter. Listed tools are auto-approved; unlisted tools are not removed, they fall through to the permission flow. On its own, allowed_tools leaves the entire built-in Claude Code toolset (including Bash, Write, Edit, and WebFetch) in the agent's context. A deal-risk agent has no business holding Bash. To remove built-ins from availability, pass tools=[]; the agent can then use only the MCP tools you registered.
Second, because the registered surface came from Step 2 (scoped by identity, then narrowed by role), the allowlist is derived from the discovered names, never typed by hand. This also sidesteps a known rough edge: the SDK historically had no clean way to allow dynamically minted MCP tool names ahead of time. Generating the list from what list_scoped_tools returned is the answer.
from claude_agent_sdk import create_sdk_mcp_server, ClaudeAgentOptions
SERVER_NAME = "deal_risk"
def build_options(scoped_tools: list[dict], identifier: str) -> ClaudeAgentOptions:
# One in-process server holds the (identity-scoped, role-narrowed) surface.
sdk_tools = [make_sdk_tool(t, identifier) for t in scoped_tools]
server = create_sdk_mcp_server(
name=SERVER_NAME,
version="1.0.0",
tools=sdk_tools,
)
# Fully qualified names: mcp__{server}__{tool}. Derived, not hardcoded.
allowed = [f"mcp__{SERVER_NAME}__{t['name']}" for t in scoped_tools]
return ClaudeAgentOptions(
mcp_servers={SERVER_NAME: server},
allowed_tools=allowed, # auto-approve exactly the scoped surface
tools=[], # strip ALL built-ins: no Bash/Write/WebFetch
model=os.environ.get("ANTHROPIC_MODEL", "claude-sonnet-4-6"),
max_turns=40,
system_prompt=DEAL_RISK_SYSTEM_PROMPT,
)
The result: this run's agent can call only the deal-risk tools this rep authorized, each executing as this rep, and nothing else exists in its context to call.
Step 5: Run the Agent
Now the objective. The scoring rubric goes in the system prompt, where reasoning belongs; the write-backs go through the same scoped, identity-bound tools, so the note that lands on a deal is authored by the rep, and Attio's own permissions apply to it.
DEAL_RISK_SYSTEM_PROMPT = """You are a deal-risk analyst for one sales rep.
Goal: surface the rep's at-risk open deals and act on them.
Steps:
1. List the rep's Gong calls from the last 3 days (use full ISO 8601 datetimes;
a date-only string returns nothing). Fetch each transcript.
2. From each transcript extract: sentiment (0.0-1.0), engagement (low/med/high),
competitor mentions, and unresolved objections.
3. Match each call to an Attio deal (search by company/email). Read stage, value,
and close date.
4. Score risk 0.0-1.0 by weighting: negative sentiment, few days to close with
low recent engagement, repeated competitor mentions, and open objections.
5. For every deal scoring >= 0.6: write a concise risk note to the Attio deal
(attio_create_note) and open one follow-up task (attio_create_task).
6. Post a single Slack message to the channel: deals ranked by risk, each with
score, one-line reason, and the agreed next step.
Never delete or change deal stage. Only create notes, create tasks, and post to Slack.
If a tool fails, skip that deal and continue.
"""
async def run_for_rep(identifier: str) -> None:
from claude_agent_sdk import ClaudeSDKClient, AssistantMessage, TextBlock, ResultMessage
# 1. Inbound identity: every connection must be ACTIVE for this rep.
if not all(ensure_authorized(c, identifier) for c in CONNECTIONS):
print("Authorize the links above, then re-run.")
return
# 2. Scoped, role-narrowed tool surface for this rep.
scoped_tools = discover_scoped_tools(identifier)
print(f"Registered {len(scoped_tools)} deal-risk tools for {identifier}")
# 3 + 4. Bind identity into the surface and contain the blast radius.
options = build_options(scoped_tools, identifier)
channel = os.environ["SLACK_CHANNEL_ID"]
prompt = f"Run today's deal-risk review and post the brief to Slack channel {channel}."
# 5. Run the autonomous loop. Claude chooses which scoped tools to call.
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) and message.subtype == "success":
print("\n[run complete]", message.result)
if __name__ == "__main__":
# In production, resolve this from the authenticated session, never client input.
rep = os.environ.get("REP_IDENTIFIER", "rep_123")
asyncio.run(run_for_rep(rep))
Swap rep_123 for the next rep and the entire surface changes underneath: a different scoped tool set, a different set of Gong calls, a different Attio pipeline, a different Slack identity, with no change to the code. That is the single-user-to-multi-rep conversion, complete. Adding a fourth signal later (a fourth connector) adds a connection name to CONNECTIONS and tool names to DEAL_RISK_TOOLS; it adds no new auth code, because the vault, scope checks, and per-action audit trail are the same for every connector.
What Breaks Without This
Each shortcut below produces a working demo and a different production failure. The failures are structural, not careless. Understanding these agent auth production anti-patterns is essential before shipping to real users.
One Attio/Gong/Slack token in .env, static @tool handlers
Works for the rep who owns the token
Second rep's agent reads and writes the first rep's pipeline; audit shows one account touching every deal
allowed_tools set, but no tools=[]
Agent uses the right tools in testing
Built-in Bash/Write/WebFetch remain in context; one wrong turn reaches outside the deal-risk surface
Full connector catalog registered, no list_scoped_tools
100+ tools degrade selection and burn tokens; agent picks wrong tools and invents parameters
Destructive Attio tools left in the surface
Never triggered in the happy path
A reasoning loop can select attio_delete_deal or a stage mutation on a confident wrong turn
Exceptions escape the handler
Passes when tokens are fresh
A single revoked connection stops the whole loop mid-run instead of degrading it
FAQs
Does list_scoped_tools enforce record-level permissions, or only which tools appear?
Only which tools appear. It scopes the surface to what the rep's connected accounts authorize. Record-level access is enforced at execution: because execute_tool runs against the rep's own token, Attio, Gong, and Slack apply their own role and record ACLs. Two layers, both required.
Can I reuse one ClaudeSDKClient across all reps to avoid warm-up cost?
No. The identity is baked into the tool handler closures and into the server you pass in ClaudeAgentOptions. A client built for one rep can only act as that rep. Reuse the process; mint the scoped surface and options per run, per rep.
Where does the rep's identifier come from in production?
From your authenticated session (session cookie, verified JWT, or database lookup), resolved server-side. Never accept it from client input; the tenant boundary depends on that value being trustworthy. For more detail, see our guide on access control for multi-tenant AI agents.
A rep revokes Gong access mid-run. What happens?
The next execute_tool for that connection fails, the handler returns is_error: True, and Claude reads the failure, skips that deal, and continues. The call fails closed; other reps' connections are unaffected; the event is in the audit trail.
Do the Gong, Attio, or Slack tokens ever reach the model context?
No. Tokens live in Scalekit's vault and are resolved server-side inside execute_tool. The agent sees tool results, never credentials. This is the same principle behind token vaults for AI agent workflows.
How is this different from pointing the SDK at Scalekit's hosted MCP endpoint?
Both are valid. The in-process pattern here gives you the tightest control over the surface (identity gate plus role gate, both in your code) and the exact point where identity binds. A hosted per-user MCP endpoint moves that surface out of your process entirely; it is the better fit when you want no server to host and per-run session tokens instead of per-run tool registration.
Next Steps to Start Building
- Create the three connections in AgentKit > Connections (Attio, Gong, Slack), each with the minimum scopes listed in the prerequisites, and copy the exact connection names into your .env.
- Install the SDKs (pip install claude-agent-sdk scalekit-sdk-python protobuf python-dotenv) and set ANTHROPIC_API_KEY.
- Run ensure_authorized for one rep first; complete the consent links it prints, then confirm all three print ACTIVE.
- Register the scoped surface, ship the run with tools=[] and a derived allowed_tools, and verify in the Scalekit audit logs that every Attio, Gong, and Slack call is attributed to that rep.
- Browse all connectors to add a fourth signal without new auth code: docs.scalekit.com/agentkit/connectors.
