Skip to content

feat:include available connectors in search/execute tool descriptions #165

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
shashi-stackone merged 8 commits into from
Apr 7, 2026
Merged

feat:include available connectors in search/execute tool descriptions #165

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
c6b2e3f
include available connectors in search/execute tool descriptions
shashi-stackone Apr 2, 2026
5300e1d
Fix th timeout and account ID issue in the SDK for the toolset.execut…
shashi-stackone Apr 7, 2026
b628ff6
Update old examples to new SDK API
shashi-stackone Apr 7, 2026
8454734
Fix the CI issue
shashi-stackone Apr 7, 2026
649b139
Address copilot review about the timeout and trim th search tool example
shashi-stackone Apr 7, 2026
8aa3656
Add working Workday Example
shashi-stackone Apr 7, 2026
e0ca81c
Add workday example in the test_examples
shashi-stackone Apr 7, 2026
04b8bd0
Add the STACKONE_API_KEY too in example
shashi-stackone Apr 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
320 changes: 31 additions & 289 deletions examples/search_tool_example.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,24 +1,15 @@
#!/usr/bin/env python

@shashi-stackone shashi-stackone Apr 7, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Removing example code as its not needed fr new API .. Most of the code not needed to avoid duplication.

"""
Example demonstrating dynamic tool discovery using search_tool.

The search tool allows AI agents to discover relevant tools based on natural language
queries without hardcoding tool names.
"""Search tool patterns: callable wrapper and config overrides.

Prerequisites:
- STACKONE_API_KEY environment variable set
- STACKONE_ACCOUNT_ID environment variable set (comma-separated for multiple)
- At least one linked account in StackOne (this example uses BambooHR)
For semantic search basics, see semantic_search_example.py.
For full agent execution, see agent_tool_search.py.

This example is runnable with the following command:
```bash
uv run examples/search_tool_example.py
```
Run with:
uv run python examples/search_tool_example.py
"""

import os
from __future__ import annotations

from stackone_ai import StackOneToolSet
import os

try:
from dotenv import load_dotenv
Expand All @@ -27,288 +18,39 @@
except ModuleNotFoundError:
pass

# Read account IDs from environment — supports comma-separated values
_account_ids = [aid.strip() for aid in os.getenv("STACKONE_ACCOUNT_ID", "").split(",") if aid.strip()]


def example_search_tool_basic():
"""Basic example of using the search tool for tool discovery"""
print("Example 1: Dynamic tool discovery\n")

# Initialize StackOne toolset
toolset = StackOneToolSet()

# Get all available tools using MCP-backed fetch_tools()
all_tools = toolset.fetch_tools(account_ids=_account_ids)
print(f"Total tools available: {len(all_tools)}")

if not all_tools:
print("No tools found. Check your linked accounts.")
return

# Get a search tool for dynamic discovery
search_tool = toolset.get_search_tool()

# Search for employee management tools — returns a Tools collection
tools = search_tool("manage employees create update list", top_k=5, account_ids=_account_ids)

print(f"Found {len(tools)} relevant tools:")
for tool in tools:
print(f" - {tool.name}: {tool.description}")

print()


def example_search_modes():
"""Comparing semantic vs local search modes.

Search config can be set at the constructor level or overridden per call:
- Constructor: StackOneToolSet(search={"method": "semantic"})
- Per-call: toolset.search_tools(query, search="local")

The search method controls which backend search_tools() uses:
- "semantic": cloud-based semantic vector search (higher accuracy for natural language)
- "local": local BM25+TF-IDF hybrid search (no network call to semantic API)
- "auto" (default): tries semantic first, falls back to local on failure
"""
print("Example 2: Semantic vs local search modes\n")

query = "manage employee time off"

# Constructor-level config — semantic search as the default for this toolset
print('Constructor config: StackOneToolSet(search={"method": "semantic"})')
toolset_semantic = StackOneToolSet(search={"method": "semantic"})
try:
tools_semantic = toolset_semantic.search_tools(query, account_ids=_account_ids, top_k=5)
print(f" Found {len(tools_semantic)} tools:")
for tool in tools_semantic:
print(f" - {tool.name}")
except Exception as e:
print(f" Semantic search unavailable: {e}")
print()

# Constructor-level config — local search (no network call to semantic API)
print('Constructor config: StackOneToolSet(search={"method": "local"})')
toolset_local = StackOneToolSet(search={"method": "local"})
tools_local = toolset_local.search_tools(query, account_ids=_account_ids, top_k=5)
print(f" Found {len(tools_local)} tools:")
for tool in tools_local:
print(f" - {tool.name}")
print()

# Per-call override — constructor defaults can be overridden on each call
print("Per-call override: constructor uses semantic, but this call uses local")
tools_override = toolset_semantic.search_tools(query, account_ids=_account_ids, top_k=5, search="local")
print(f" Found {len(tools_override)} tools:")
for tool in tools_override:
print(f" - {tool.name}")
print()

# Auto (default) — tries semantic, falls back to local
print('Default: StackOneToolSet() uses search="auto" (semantic with local fallback)')
toolset_auto = StackOneToolSet()
tools_auto = toolset_auto.search_tools(query, account_ids=_account_ids, top_k=5)
print(f" Found {len(tools_auto)} tools:")
for tool in tools_auto:
print(f" - {tool.name}")
print()


def example_top_k_config():
"""Configuring top_k at the constructor level vs per-call.

Constructor-level top_k applies to all search_tools() and search_action_names()
calls. Per-call top_k overrides the constructor default for that single call.
"""
print("Example 3: top_k at constructor vs per-call\n")

# Constructor-level top_k — all calls default to returning 3 results
toolset = StackOneToolSet(search={"top_k": 3})

query = "manage employee records"
print(f'Constructor top_k=3: searching for "{query}"')
tools_default = toolset.search_tools(query, account_ids=_account_ids)
print(f" Got {len(tools_default)} tools (constructor default)")
for tool in tools_default:
print(f" - {tool.name}")
print()

# Per-call override — this single call returns up to 10 results
print("Per-call top_k=10: overriding constructor default")
tools_override = toolset.search_tools(query, account_ids=_account_ids, top_k=10)
print(f" Got {len(tools_override)} tools (per-call override)")
for tool in tools_override:
print(f" - {tool.name}")
print()


def example_search_tool_with_execution():
"""Example of discovering and executing tools dynamically"""
print("Example 4: Dynamic tool execution\n")

# Initialize toolset
toolset = StackOneToolSet()

# Get all tools using MCP-backed fetch_tools()
all_tools = toolset.fetch_tools(account_ids=_account_ids)

if not all_tools:
print("No tools found. Check your linked accounts.")
return

search_tool = toolset.get_search_tool()

# Step 1: Search for relevant tools
tools = search_tool("list all employees", top_k=1, account_ids=_account_ids)

if tools:
best_tool = tools[0]
print(f"Best matching tool: {best_tool.name}")
print(f"Description: {best_tool.description}")

# Step 2: Execute the found tool directly
try:
print(f"\nExecuting {best_tool.name}...")
result = best_tool(limit=5)
print(f"Execution result: {result}")
except Exception as e:
print(f"Execution failed (expected in example): {e}")

print()


def example_with_openai():
"""Example of using search tool with OpenAI"""
print("Example 5: Using search tool with OpenAI\n")

try:
from openai import OpenAI

# Initialize OpenAI client
client = OpenAI()

# Initialize StackOne toolset
toolset = StackOneToolSet()

# Search for BambooHR employee tools
tools = toolset.search_tools("manage employees", account_ids=_account_ids, top_k=5)

# Convert to OpenAI format
openai_tools = tools.to_openai()

# Create a chat completion with discovered tools
response = client.chat.completions.create(
model="gpt-5.4",
messages=[
{
"role": "system",
"content": "You are an HR assistant with access to employee management tools.",
},
{"role": "user", "content": "Can you help me find tools for managing employee records?"},
],
tools=openai_tools,
tool_choice="auto",
)

print("OpenAI Response:", response.choices[0].message.content)

if response.choices[0].message.tool_calls:
print("\nTool calls made:")
for tool_call in response.choices[0].message.tool_calls:
print(f" - {tool_call.function.name}")

except ImportError:
print("OpenAI library not installed. Install with: pip install openai")
except Exception as e:
print(f"OpenAI example failed: {e}")

print()


def example_with_langchain():
"""Example of using tools with LangChain"""
print("Example 6: Using tools with LangChain\n")

try:
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

# Initialize StackOne toolset
toolset = StackOneToolSet()

# Get tools and convert to LangChain format using MCP-backed fetch_tools()
tools = toolset.search_tools("list employees", account_ids=_account_ids, top_k=5)
langchain_tools = list(tools.to_langchain())

print(f"Available tools for LangChain: {len(langchain_tools)}")
for tool in langchain_tools:
print(f" - {tool.name}: {tool.description}")

# Create LangChain agent
llm = ChatOpenAI(model="gpt-5.4", temperature=0)

prompt = ChatPromptTemplate.from_messages(
[
(
"system",
"You are an HR assistant. Use the available tools to help the user.",
),
("human", "{input}"),
("placeholder", "{agent_scratchpad}"),
]
)

agent = create_tool_calling_agent(llm, langchain_tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=langchain_tools, verbose=True)

# Run the agent
result = agent_executor.invoke({"input": "Find tools that can list employee data"})
from stackone_ai import StackOneToolSet

print(f"\nAgent result: {result['output']}")
account_id = os.getenv("STACKONE_ACCOUNT_ID", "")
_account_ids = [a.strip() for a in account_id.split(",") if a.strip()] if account_id else []

except ImportError as e:
print(f"LangChain dependencies not installed: {e}")
print("Install with: pip install langchain-openai")
except Exception as e:
print(f"LangChain example failed: {e}")

print()
# --- Example 1: get_search_tool() callable ---
print("=== get_search_tool() callable ===\n")

toolset = StackOneToolSet(search={})
Comment thread
cubic-dev-ai[bot] marked this conversation as resolved.
Outdated
search_tool = toolset.get_search_tool()

def main():
"""Run all examples"""
print("=" * 60)
print("StackOne AI SDK - Search Tool Examples")
print("=" * 60)
print()
queries = ["cancel an event", "list employees", "send a message"]
for query in queries:
tools = search_tool(query, top_k=3, account_ids=_account_ids)
names = [t.name for t in tools]
print(f' "{query}" -> {", ".join(names) or "(none)"}')

if not os.getenv("STACKONE_API_KEY"):
print("Set STACKONE_API_KEY to run these examples.")
return

if not _account_ids:
print("Set STACKONE_ACCOUNT_ID to run these examples.")
print("(Comma-separated for multiple accounts)")
return
# --- Example 2: Constructor top_k vs per-call override ---
print("\n=== Constructor top_k vs per-call override ===\n")

# Basic examples that work without external APIs
example_search_tool_basic()
example_search_modes()
example_top_k_config()
example_search_tool_with_execution()
toolset_3 = StackOneToolSet(search={"top_k": 3})
toolset_10 = StackOneToolSet(search={"top_k": 10})

# Examples that require OpenAI API
if os.getenv("OPENAI_API_KEY"):
example_with_openai()
example_with_langchain()
else:
print("Set OPENAI_API_KEY to run OpenAI and LangChain examples\n")
query = "manage employee records"

print("=" * 60)
print("Examples completed!")
print("=" * 60)
tools_3 = toolset_3.search_tools(query, account_ids=_account_ids)
print(f"Constructor top_k=3: got {len(tools_3)} tools")

tools_10 = toolset_10.search_tools(query, account_ids=_account_ids)
print(f"Constructor top_k=10: got {len(tools_10)} tools")

if __name__ == "__main__":
main()
# Per-call override: constructor says 3 but this call says 10
tools_override = toolset_3.search_tools(query, top_k=10, account_ids=_account_ids)
print(f"Per-call top_k=10 (overrides constructor 3): got {len(tools_override)} tools")
1 change: 0 additions & 1 deletion examples/semantic_search_example.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -133,7 +133,6 @@ def example_search_action_names():
print(f"Top {len(results_limited)} matches from the full catalog:")
for r in results_limited:
print(f" [{r.similarity_score:.2f}] {r.id}")
print(f" {r.description}")

@shashi-stackone shashi-stackone Apr 7, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not aligned to new SDK API

print()

# Show filtering effect when account_ids are available
Expand Down
3 changes: 2 additions & 1 deletion stackone_ai/models.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,7 @@ class ExecuteConfig(BaseModel):
parameter_locations: dict[str, ParameterLocation] = Field(
default_factory=dict, description="Maps parameter names to their location in the request"
)
timeout: float = Field(default=60.0, description="Request timeout in seconds")


class ToolParameters(BaseModel):
Expand Down Expand Up @@ -249,7 +250,7 @@ def execute(
if query_params:
request_kwargs["params"] = query_params

response = httpx.request(**request_kwargs)
response = httpx.request(**request_kwargs, timeout=self._execute_config.timeout)
response_status = response.status_code
response.raise_for_status()

Expand Down
Loading
Loading