Overview
The ToolKitClient class allows you to register Python functions as tools and serve them via WebSocket. It handles:
- Tool registration with metadata
- WebSocket(WS(S))/Hyper Text Transfer Protocol(HTTP(S)) connection to the ATP server
- Tool invocation with parameter validation
- OAuth2 token injection for authenticated tools
- Automatic reconnection on connection loss
Constructor
from atp_sdk.clients import ToolKitClient
client = ToolKitClient(
api_key: str,
app_name: str,
base_url: str = "https://api.chat-atp.com"
)
Parameters
Name of your application/toolkit. This will be used to identify your toolkit on the ATP platform.
base_url
string
default:"https://api.chat-atp.com.com"
ATP Server backend URL. Use the default unless you’re running a custom ATP server.
Methods
Registers a Python function as a tool with the ATP platform.
@client.register_tool(
function_name: str,
params: list[str],
required_params: list[str],
description: str,
auth_provider: Optional[str] = None,
auth_type: Optional[str] = None,
auth_with: Optional[str] = None
)
def my_tool(**kwargs):
...
Parameters
Unique name for the tool. This is how the LLM will reference the tool.
List of all parameter names the tool accepts.
List of required parameter names. The tool will fail if these are not provided.
Human-readable description of what the tool does. This helps the LLM understand when to use the tool.
Name of OAuth2 provider (e.g., “hubspot”, “google”, “salesforce”). Set to None if no authentication is required.
Authentication type (e.g., “OAuth2”, “apiKey”). Set to None if no authentication is required.
Name of the token parameter (e.g., “access_token”, “api_key”). Set to None if no authentication is required.
Returns
A decorator to wrap your function.
start
Starts the WebSocket client and begins listening for tool requests.
This method blocks the main thread. Make sure all tools are registered before calling start().
stop
Stops the WebSocket client and closes the connection.
Your tool functions must follow these requirements:
- Accept
**kwargs: All parameters must be passed as keyword arguments.
- Return serializable data: Return a dict, str, list, or other JSON-serializable type.
- Handle
auth_token: If your tool requires authentication, expect auth_token in kwargs.
def my_tool(**kwargs):
# Get parameters
param1 = kwargs.get('param1')
param2 = kwargs.get('param2')
# Get auth token (if required)
auth_token = kwargs.get('auth_token')
# Do something
result = do_something(param1, param2, auth_token)
# Return serializable data
return {"result": result}
Examples
from atp_sdk.clients import ToolKitClient
client = ToolKitClient(
api_key="YOUR_ATP_TOOLKIT_API_KEY",
app_name="my_app"
)
@client.register_tool(
function_name="echo",
params=['text'],
required_params=['text'],
description="Echoes the input text.",
auth_provider=None,
auth_type=None,
auth_with=None
)
def echo(**kwargs):
return {"echo": kwargs.get('text')}
client.start()
import requests
from atp_sdk.clients import ToolKitClient
client = ToolKitClient(
api_key="YOUR_ATP_TOOLKIT_API_KEY",
app_name="hubspot_toolkit"
)
@client.register_tool(
function_name="create_company",
params=['name', 'domain', 'industry'],
required_params=['name', 'domain', 'industry'],
description="Creates a company in HubSpot.",
auth_provider="hubspot",
auth_type="OAuth2",
auth_with="access_token"
)
def create_company(**kwargs):
access_token = kwargs.get('auth_token')
url = "https://api.hubapi.com/crm/v3/objects/companies"
headers = {
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json"
}
data = {
"properties": {
"name": kwargs.get('name'),
"domain": kwargs.get('domain'),
"industry": kwargs.get('industry')
}
}
response = requests.post(url, json=data, headers=headers)
return response.json()
client.start()
You can register multiple tools with the same client:
from atp_sdk.clients import ToolKitClient
client = ToolKitClient(
api_key="YOUR_ATP_TOOLKIT_API_KEY",
app_name="my_toolkit"
)
@client.register_tool(
function_name="tool1",
params=['param1'],
required_params=['param1'],
description="First tool.",
auth_provider=None,
auth_type=None,
auth_with=None
)
def tool1(**kwargs):
return {"result": "tool1"}
@client.register_tool(
function_name="tool2",
params=['param2'],
required_params=['param2'],
description="Second tool.",
auth_provider=None,
auth_type=None,
auth_with=None
)
def tool2(**kwargs):
return {"result": "tool2"}
client.start()
Error Handling
The ATP SDK automatically handles errors and returns them in a standardized format:
Missing Required Parameters
{
"error": "Missing required parameter: name"
}
Missing Auth Token
{
"error": "Missing auth_token for authenticated tool"
}
Function Exception
{
"error": "Error message from exception"
}
WebSocket Events
When you call client.start(), all registered tools are announced to the ATP backend.
When a tool request is received, your function is called with the provided parameters and (if needed) auth_token.
Example incoming message:
{
"message_type": "atp_tool_request",
"payload": {
"request_id": "uuid",
"tool_name": "create_company",
"params": {
"name": "Acme",
"domain": "acme.com",
"industry": "Tech"
},
"auth_token": "ACCESS_TOKEN"
}
}
Advanced Usage
Custom Backend
If you’re running a custom ATP server, you can specify the base URL:
client = ToolKitClient(
api_key="YOUR_API_KEY",
app_name="my_app",
base_url="https://your-backend.example.com"
)
Next Steps
