| title | Python SDK reference | |||||
|---|---|---|---|---|---|---|
| description | Complete API reference for the Scalekit Python SDK: actions client, MCP server provisioning, framework adapters, tools client, and modifiers. | |||||
| sidebar |
|
|||||
| tableOfContents |
|
|||||
| head |
|
import { Aside } from '@astrojs/starlight/components'; import MethodParams from '@/components/MethodParams.astro'; import MethodReturns from '@/components/MethodReturns.astro';
scalekit_client.actions is the primary interface for AgentKit. It handles connected account management, MCP server provisioning, tool execution, and framework integrations.
pip install scalekit-sdk-pythonimport os
import scalekit.client
scalekit_client = scalekit.client.ScalekitClient(
client_id=os.getenv("SCALEKIT_CLIENT_ID"),
client_secret=os.getenv("SCALEKIT_CLIENT_SECRET"),
env_url=os.getenv("SCALEKIT_ENV_URL"),
)
actions = scalekit_client.actionsGenerates a time-limited OAuth magic link to authorize a user's connection.
<MethodParams label="Input schema" params={[ { name: 'identifier', type: 'str', required: false, description: 'User identifier (e.g. email)' }, { name: 'connection_name', type: 'str', required: false, description: 'Connector slug (e.g. gmail)' }, { name: 'connected_account_id', type: 'str', required: false, description: 'Direct connected account ID (ca_...)' }, { name: 'state', type: 'str', required: false, description: 'Opaque value passed through to the redirect URL' }, { name: 'user_verify_url', type: 'str', required: false, description: 'App redirect URL for user verification' }, ]} />
<MethodReturns type="MagicLinkResponse" fields={[ { name: 'link', type: 'str', description: 'OAuth magic link URL. Redirect the user here to start the authorization flow.' }, { name: 'expiry', type: 'datetime', description: 'Link expiry timestamp' }, ]} />
magic_link = actions.get_authorization_link(
identifier="user@example.com",
connection_name="gmail",
user_verify_url="https://your-app.com/verify",
)
# Redirect the user to magic_link.linkVerifies the user after OAuth callback. Call this from your redirect URL handler.
<MethodParams label="Input schema" params={[ { name: 'auth_request_id', type: 'str', required: true, description: 'Token from the redirect URL query params' }, { name: 'identifier', type: 'str', required: true, description: 'Current user identifier' }, ]} />
<MethodReturns type="VerifyConnectedAccountUserResponse" fields={[ { name: 'post_user_verify_redirect_url', type: 'str', description: 'URL to redirect the user to after successful verification' }, ]} />
result = actions.verify_connected_account_user(
auth_request_id=request.args["auth_request_id"],
identifier="user@example.com",
)
# Redirect to result.post_user_verify_redirect_urlFetches an existing connected account or creates one if none exists. Use this as the default when setting up a user.
<MethodParams label="Input schema" params={[ { name: 'connection_name', type: 'str', required: true, description: 'Connector slug' }, { name: 'identifier', type: 'str', required: true, description: "User's identifier" }, { name: 'authorization_details', type: 'dict', required: false, description: 'OAuth token or static auth details' }, { name: 'organization_id', type: 'str', required: false, description: 'Organization tenant ID when your app scopes auth and accounts by org' }, { name: 'user_id', type: 'str', required: false, description: 'Your application user ID when you map Scalekit accounts to internal users' }, { name: 'api_config', type: 'dict', required: false, description: 'Connector-specific options (for example scopes or static auth fields)' }, ]} />
<MethodReturns type="CreateConnectedAccountResponse" fields={[ { name: 'connected_account.id', type: 'str', description: 'Account ID (ca_...)' }, { name: 'connected_account.identifier', type: 'str', description: "User's identifier" }, { name: 'connected_account.provider', type: 'str', description: 'Provider slug' }, { name: 'connected_account.status', type: 'str', description: 'ACTIVE, INACTIVE, or PENDING' }, { name: 'connected_account.authorization_type', type: 'str', description: 'OAuth, API_KEY, etc.' }, { name: 'connected_account.token_expires_at', type: 'datetime', description: 'OAuth token expiry' }, ]} />
account = actions.get_or_create_connected_account(
connection_name="gmail",
identifier="user@example.com",
)
print(account.connected_account.id)Fetches auth details for a connected account. Returns sensitive credentials. Protect access to this method.
Use this when you know the connected account already exists and you need its credential payload. For first-time setup or general application flows, prefer get_or_create_connected_account so new users do not hit a not-found error.
Requires connected_account_id or connection_name + identifier.
<MethodParams label="Input schema" params={[ { name: 'connection_name', type: 'str', required: false, description: 'Connector slug. Use with identifier when you do not pass connected_account_id.' }, { name: 'identifier', type: 'str', required: false, description: 'End-user or workspace identifier. Use with connection_name.' }, { name: 'connected_account_id', type: 'str', required: false, description: 'Connected account ID (ca_...) when resolving by ID instead of name + identifier' }, ]} />
<MethodReturns type="GetConnectedAccountAuthResponse" fields={[ { name: 'connected_account.id', type: 'str', description: 'Account ID (ca_...)' }, { name: 'connected_account.identifier', type: 'str', description: "User's identifier" }, { name: 'connected_account.provider', type: 'str', description: 'Provider slug' }, { name: 'connected_account.status', type: 'str', description: 'ACTIVE, INACTIVE, or PENDING' }, { name: 'connected_account.authorization_type', type: 'str', description: 'OAuth, API_KEY, etc.' }, { name: 'connected_account.authorization_details', type: 'dict', description: 'Credential payload (access token, API key, etc.)' }, { name: 'connected_account.token_expires_at', type: 'datetime', description: 'OAuth token expiry' }, { name: 'connected_account.last_used_at', type: 'datetime', description: 'Last time this account was used' }, { name: 'connected_account.updated_at', type: 'datetime', description: 'Last update timestamp' }, ]} />
<MethodParams label="Input schema" params={[ { name: 'connection_name', type: 'str', required: false, description: 'Filter by connector' }, { name: 'identifier', type: 'str', required: false, description: 'Filter by user identifier' }, { name: 'provider', type: 'str', required: false, description: 'Filter by provider' }, ]} />
<MethodReturns type="ListConnectedAccountsResponse" fields={[ { name: 'connected_accounts', type: 'list', description: 'List of ConnectedAccountForList objects (excludes authorization_details and api_config)' }, { name: 'total_count', type: 'int', description: 'Total number of matching accounts' }, { name: 'next_page_token', type: 'str', description: 'Token for the next page, if any' }, { name: 'previous_page_token', type: 'str', description: 'Token for the previous page, if any' }, ]} />
Creates a connected account with explicit auth details.
<MethodParams label="Input schema" params={[ { name: 'connection_name', type: 'str', required: true, description: 'Connector slug. Must match a connection configured in your environment.' }, { name: 'identifier', type: 'str', required: true, description: 'Stable ID for this end user or workspace (email, user_id, or custom string)' }, { name: 'authorization_details', type: 'dict', required: true, description: 'OAuth token payload, API key, or other credentials for this connector' }, { name: 'organization_id', type: 'str', required: false, description: 'Organization tenant ID when your app scopes auth and accounts by org' }, { name: 'user_id', type: 'str', required: false, description: 'Your application user ID when you map Scalekit accounts to internal users' }, { name: 'api_config', type: 'dict', required: false, description: 'Connector-specific options (for example scopes or static auth fields)' }, ]} />
Returns CreateConnectedAccountResponse. Same shape as get_or_create_connected_account.
Requires connected_account_id or connection_name + identifier.
<MethodParams label="Input schema" params={[ { name: 'connection_name', type: 'str', required: false, description: 'Connector slug. Use with identifier when you do not pass connected_account_id.' }, { name: 'identifier', type: 'str', required: false, description: 'End-user or workspace identifier. Use with connection_name.' }, { name: 'connected_account_id', type: 'str', required: false, description: 'Connected account ID (ca_...) when updating by ID instead of name + identifier' }, { name: 'authorization_details', type: 'dict', required: false, description: 'Replace or merge stored credentials (OAuth tokens, API keys, etc.)' }, { name: 'organization_id', type: 'str', required: false, description: 'Organization tenant ID when your app scopes auth and accounts by org' }, { name: 'user_id', type: 'str', required: false, description: 'Your application user ID when you map Scalekit accounts to internal users' }, { name: 'api_config', type: 'dict', required: false, description: 'Connector-specific configuration to persist on the account' }, ]} />
Returns UpdateConnectedAccountResponse.
Deletes a connected account and revokes its credentials. Requires connected_account_id or connection_name + identifier.
<MethodParams label="Input schema" params={[ { name: 'connection_name', type: 'str', required: false, description: 'Connector slug. Use with identifier when you do not pass connected_account_id.' }, { name: 'identifier', type: 'str', required: false, description: 'End-user or workspace identifier. Use with connection_name.' }, { name: 'connected_account_id', type: 'str', required: false, description: 'Connected account ID (ca_...) when deleting by ID instead of name + identifier' }, ]} />
Returns DeleteConnectedAccountResponse.
Executes a named tool via Scalekit. Pre- and post-modifiers run automatically if registered.
<MethodParams label="Input schema" params={[ { name: 'tool_name', type: 'str', required: true, description: 'Tool name (e.g. gmail_fetch_emails)' }, { name: 'tool_input', type: 'dict', required: true, description: 'Parameters the tool expects' }, { name: 'identifier', type: 'str', required: false, description: "User's identifier" }, { name: 'connected_account_id', type: 'str', required: false, description: 'Direct connected account ID' }, ]} />
<MethodReturns type="ExecuteToolResponse" fields={[ { name: 'data', type: 'dict', description: 'Tool structured output' }, { name: 'execution_id', type: 'str', description: 'Unique ID for this execution' }, ]} />
result = actions.execute_tool(
tool_name="gmail_fetch_emails",
tool_input={"max_results": 5, "label": "UNREAD"},
identifier="user@example.com",
)
emails = result.dataMakes a REST API call on behalf of a connected account. Scalekit injects the user's OAuth token automatically.
<MethodParams label="Input schema" params={[ { name: 'connection_name', type: 'str', required: true, description: 'Connector slug' }, { name: 'identifier', type: 'str', required: true, description: "User's identifier" }, { name: 'path', type: 'str', required: true, description: 'API path (e.g. /gmail/v1/users/me/messages)' }, { name: 'method', type: 'str', required: false, description: 'HTTP method. Default: GET' }, { name: 'query_params', type: 'dict', required: false, description: 'URL query parameters appended to path' }, { name: 'body', type: 'any', required: false, description: 'JSON-serializable body for POST, PUT, PATCH, or similar methods' }, { name: 'form_data', type: 'dict', required: false, description: 'Multipart form fields when the upstream API expects form data instead of JSON' }, { name: 'headers', type: 'dict', required: false, description: 'Extra HTTP headers merged with Scalekit-injected auth headers' }, ]} />
Returns requests.Response. Use .json(), .status_code, and standard response attributes.
response = actions.request(
connection_name="gmail",
identifier="user@example.com",
path="/gmail/v1/users/me/messages",
query_params={"maxResults": 5, "q": "is:unread"},
)
messages = response.json()["messages"]actions.mcp generates per-user MCP-compatible server URLs. Any MCP-compatible agent framework (LangChain, Google ADK, Anthropic, OpenAI, and others) can connect to these URLs directly.
Two-step model: Create a config once (defines which connectors and tools to expose), then call ensure_instance per user to get their personal MCP server URL.
<MethodParams label="Input schema" params={[ { name: 'name', type: 'str', required: true, description: 'Config name' }, { name: 'description', type: 'str', required: false, description: 'Human-readable summary of what this MCP config exposes' }, { name: 'connection_tool_mappings', type: 'list', required: false, description: 'List of McpConfigConnectionToolMapping objects' }, ]} />
<MethodReturns type="CreateMcpConfigResponse" fields={[ { name: 'config.id', type: 'str', description: 'Config ID' }, { name: 'config.name', type: 'str', description: 'Config name' }, { name: 'config.connection_tool_mappings', type: 'list', description: 'Connector-to-tools mappings' }, ]} />
from scalekit.actions.types import McpConfigConnectionToolMapping
config = actions.mcp.create_config(
name="email-agent",
connection_tool_mappings=[
McpConfigConnectionToolMapping(
connection_name="gmail",
tools=["gmail_fetch_emails", "gmail_send_email"],
)
],
)<MethodParams label="Input schema" params={[ { name: 'page_size', type: 'int', required: false, description: 'Maximum configs per page (server default if omitted)' }, { name: 'page_token', type: 'str', required: false, description: 'Opaque cursor from a previous list response' }, { name: 'filter_name', type: 'str', required: false, description: 'Filter by exact name' }, { name: 'filter_provider', type: 'str', required: false, description: 'Filter by provider slug' }, { name: 'search', type: 'str', required: false, description: 'Free-text search on name' }, ]} />
Returns ListMcpConfigsResponse.
<MethodParams label="Input schema" params={[ { name: 'config_id', type: 'str', required: true, description: 'MCP config ID from create_config or list_configs' }, { name: 'description', type: 'str', required: false, description: 'New human-readable description for this config' }, { name: 'connection_tool_mappings', type: 'list', required: false, description: 'Replaces existing mappings' }, ]} />
Returns UpdateMcpConfigResponse.
<MethodParams label="Input schema" params={[ { name: 'config_id', type: 'str', required: true, description: 'MCP config ID to delete' }, ]} />
Returns DeleteMcpConfigResponse.
Creates an MCP instance for this user if one doesn't exist, or returns the existing one. Call this on every session; it's idempotent.
The instance.url field is the MCP server URL to give to the user's agent or IDE.
<MethodParams label="Input schema" params={[ { name: 'config_name', type: 'str', required: true, description: 'Name of the config to instantiate' }, { name: 'user_identifier', type: 'str', required: true, description: 'User identifier (e.g. email)' }, { name: 'name', type: 'str', required: false, description: 'Display name for the instance' }, ]} />
<MethodReturns type="EnsureMcpInstanceResponse" fields={[ { name: 'instance.url', type: 'str', description: 'MCP server URL for agent or IDE' }, { name: 'instance.id', type: 'str', description: 'Instance ID' }, { name: 'instance.name', type: 'str', description: 'Display name' }, { name: 'instance.user_identifier', type: 'str', description: 'User identifier' }, { name: 'instance.config', type: 'object', description: 'The config this instance was created from' }, { name: 'instance.last_used_at', type: 'datetime', description: 'Last usage timestamp' }, { name: 'instance.updated_at', type: 'datetime', description: 'Last update timestamp' }, ]} />
instance = actions.mcp.ensure_instance(
config_name="email-agent",
user_identifier="user@example.com",
)
mcp_url = instance.instance.url
# Give mcp_url to the user's agent or IDEReturns authorization status per connector. Use include_auth_links=True to generate fresh auth links for connections that need authorization or re-authorization.
<MethodParams label="Input schema" params={[ { name: 'instance_id', type: 'str', required: true, description: 'Instance ID' }, { name: 'include_auth_links', type: 'bool', required: false, description: 'Generate auth links for unauthorized connections' }, ]} />
<MethodReturns type="GetMcpInstanceAuthStateResponse" fields={[ { name: 'connections', type: 'list', description: 'List of McpInstanceConnectionAuthState, one per configured connector' }, { name: 'connections[].connection_id', type: 'str', description: 'Connection ID' }, { name: 'connections[].connection_name', type: 'str', description: 'Connector slug' }, { name: 'connections[].provider', type: 'str', description: 'Provider slug' }, { name: 'connections[].connected_account_id', type: 'str', description: 'Connected account ID, if authorized' }, { name: 'connections[].connected_account_status', type: 'str', description: 'ACTIVE, INACTIVE, or PENDING' }, { name: 'connections[].authentication_link', type: 'str', description: 'Auth link to send to the user when status is not ACTIVE' }, ]} />
auth_state = actions.mcp.get_instance_auth_state(
instance_id=instance.instance.id,
include_auth_links=True,
)
for conn in auth_state.connections:
if conn.connected_account_status != "ACTIVE":
# Send conn.authentication_link to the user to authorize
print(f"{conn.connection_name}: {conn.authentication_link}")<MethodParams label="Input schema" params={[ { name: 'instance_id', type: 'str', required: true, description: 'MCP instance ID from ensure_instance or list_instances' }, ]} />
Returns GetMcpInstanceResponse.
<MethodParams label="Input schema" params={[ { name: 'page_size', type: 'int', required: false, description: 'Maximum instances per page (server default if omitted)' }, { name: 'page_token', type: 'str', required: false, description: 'Opaque cursor from a previous list response' }, { name: 'filter_user_identifier', type: 'str', required: false, description: 'Filter by user' }, { name: 'filter_config_name', type: 'str', required: false, description: 'Filter by config name' }, { name: 'filter_name', type: 'str', required: false, description: 'Filter by MCP instance display name' }, { name: 'filter_id', type: 'str', required: false, description: 'Filter by MCP instance ID' }, ]} />
Returns ListMcpInstancesResponse.
At least one of name or config_name is required.
<MethodParams label="Input schema" params={[ { name: 'instance_id', type: 'str', required: true, description: 'MCP instance ID to update' }, { name: 'name', type: 'str', required: false, description: 'New display name for this instance' }, { name: 'config_name', type: 'str', required: false, description: 'Switch the instance to a different config by name' }, ]} />
Returns UpdateMcpInstanceResponse.
<MethodParams label="Input schema" params={[ { name: 'instance_id', type: 'str', required: true, description: 'MCP instance ID to delete' }, ]} />
Returns DeleteMcpInstanceResponse.
Pre-built integrations for LangChain and Google ADK. Use these when your agent runs in one of these frameworks and you prefer native tool objects over an MCP URL.
`actions.mcp.ensure_instance` generates a URL compatible with any MCP-supporting framework. Use framework adapters only when native tool objects are required.pip install langchain<MethodParams label="Input schema" params={[ { name: 'identifier', type: 'str', required: true, description: 'User connected account identifier' }, { name: 'providers', type: 'list', required: false, description: 'Filter by provider (e.g. ["google"])' }, { name: 'tool_names', type: 'list', required: false, description: 'Filter by tool name' }, { name: 'connection_names', type: 'list', required: false, description: 'Filter by connection name' }, { name: 'page_size', type: 'int', required: false, description: 'Maximum tools per page (server default if omitted)' }, { name: 'page_token', type: 'str', required: false, description: 'Opaque cursor from a previous list response' }, ]} />
<MethodReturns type="List[StructuredTool]" fields={[ { name: '[].name', type: 'str', description: 'Tool name' }, { name: '[].description', type: 'str', description: 'Tool description' }, { name: '[].args_schema', type: 'object', description: 'Pydantic schema for the tool inputs' }, ]} />
from langchain.agents import create_react_agent
tools = actions.langchain.get_tools(identifier="user@example.com")
agent = create_react_agent(llm=llm, tools=tools, prompt=prompt)pip install google-adkSame parameters as actions.langchain.get_tools.
Returns List[ScalekitGoogleAdkTool]. Pass it directly to a Google ADK agent.
tools = actions.google.get_tools(identifier="user@example.com")scalekit_client.actions.tools gives access to raw tool schemas. Use this when building a custom adapter or passing schemas directly to an LLM API (e.g. Anthropic, OpenAI).
<MethodParams label="Input schema" params={[ { name: 'filter', type: 'Filter', required: false, description: 'Filter by provider, identifier, or tool name' }, { name: 'page_size', type: 'int', required: false, description: 'Maximum tools per page (server default if omitted)' }, { name: 'page_token', type: 'str', required: false, description: 'Opaque cursor from a previous list response' }, ]} />
<MethodReturns type="ListToolsResponse" fields={[ { name: 'tools', type: 'list', description: 'List of tool schemas (name, description, input schema)' }, { name: 'next_page_token', type: 'str', description: 'Token for the next page, if any' }, ]} />
Lists tools scoped to a specific user. This is what framework adapters use internally to fetch per-user tool schemas.
<MethodParams label="Input schema" params={[ { name: 'identifier', type: 'str', required: true, description: 'User connected account identifier' }, { name: 'filter', type: 'ScopedToolFilter', required: false, description: 'Filter by providers, tool names, or connection names' }, { name: 'page_size', type: 'int', required: false, description: 'Maximum tools per page (server default if omitted)' }, { name: 'page_token', type: 'str', required: false, description: 'Opaque cursor from a previous list response' }, ]} />
<MethodReturns type="ListScopedToolsResponse" fields={[ { name: 'tools', type: 'list', description: 'List of tool schemas (name, description, input_schema)' }, { name: 'tools[].name', type: 'str', description: 'Tool name' }, { name: 'tools[].description', type: 'str', description: 'Tool description' }, { name: 'tools[].input_schema', type: 'object', description: 'JSON Schema for tool inputs. Pass directly to LLM API.' }, { name: 'next_page_token', type: 'str', description: 'Token for the next page, if any' }, ]} />
tools_response = scalekit_client.actions.tools.list_scoped_tools(
identifier="user@example.com",
)
# Pass tools_response.tools to your LLM's tool call APILow-level tool execution. Bypasses modifiers. Prefer actions.execute_tool in most cases.
<MethodParams label="Input schema" params={[ { name: 'tool_name', type: 'str', required: true, description: 'Registered tool name to execute' }, { name: 'identifier', type: 'str', required: true, description: 'End-user or workspace identifier used to resolve the connected account' }, { name: 'params', type: 'dict', required: false, description: 'Tool arguments matching the tool input schema' }, { name: 'connected_account_id', type: 'str', required: false, description: 'Connected account ID (ca_...) when you already know it' }, ]} />
Returns ExecuteToolResponse. Same shape as actions.execute_tool.
Modifiers intercept tool calls to transform inputs or outputs, useful for validation, enrichment, or logging.
@actions.pre_modifier(tool_names=["gmail_fetch_emails"])
def add_default_label(tool_input):
tool_input.setdefault("label", "UNREAD")
return tool_input
@actions.post_modifier(tool_names=["gmail_fetch_emails"])
def filter_attachments(tool_output):
tool_output["emails"] = [e for e in tool_output["emails"] if not e.get("has_attachment")]
return tool_output| Decorator | Receives | Returns |
|---|---|---|
@actions.pre_modifier(tool_names) |
dict |
Modified dict |
@actions.post_modifier(tool_names) |
dict |
Modified dict |
tool_names accepts a string or a list of strings. Multiple modifiers for the same tool chain in registration order.
from scalekit.common.exceptions import ScalekitNotFoundException, ScalekitServerException
try:
account = actions.get_connected_account(
connection_name="gmail",
identifier="user@example.com",
)
except ScalekitNotFoundException:
# Account does not exist: create it or redirect to auth
pass
except ScalekitServerException as e:
print(e.error_code, e.http_status)| Exception | When raised |
|---|---|
ScalekitNotFoundException |
Resource not found |
ScalekitUnauthorizedException |
Invalid credentials |
ScalekitForbiddenException |
Insufficient permissions |
ScalekitServerException |
Base class for all server errors |