From 62e838d200eaf47aa072f8ac36b254649d09a4cb Mon Sep 17 00:00:00 2001 From: ryoppippi <1560508+ryoppippi@users.noreply.github.com> Date: Tue, 11 Nov 2025 15:50:54 +0000 Subject: [PATCH 1/3] feat: implement MCP-backed fetch_tools() with dynamic tool discovery Add MCP (Model Context Protocol) integration to enable dynamic tool fetching at runtime, matching the Node SDK functionality. This allows users to pull the latest tool definitions from the StackOne API without SDK updates. Key Features: - Dynamic tool discovery via MCP server endpoint - _StackOneRpcTool class for RPC-backed execution at /actions/rpc - Account filtering with set_accounts() and account_ids parameter - Provider filtering (case-insensitive, prefix-based) - Action filtering with glob pattern support - Thread-safe async execution wrapper (_run_async) - Proper authentication with Basic Auth headers Implementation Details: - MCP client integration with streamablehttp transport - Tool catalog fetching with pagination support (cursor-based) - Schema normalization for proper nullable field handling - RPC payload construction with body/headers/path/query support - Account scoping via x-account-id header - User-Agent tracking for SDK version telemetry Dependencies: - Requires optional 'mcp' extra: pip install 'stackone-ai[mcp]' - Raises ToolsetConfigError if MCP dependencies not available The implementation maintains API parity with the Node SDK's stackone.mcp-fetch functionality while following Python idioms. --- stackone_ai/toolset.py | 281 ++++++++++++++++++++++++++++++++++++++--- 1 file changed, 262 insertions(+), 19 deletions(-) diff --git a/stackone_ai/toolset.py b/stackone_ai/toolset.py index e5b2235..008d05d 100644 --- a/stackone_ai/toolset.py +++ b/stackone_ai/toolset.py @@ -1,18 +1,52 @@ # TODO: Remove when Python 3.9 support is dropped from __future__ import annotations +import asyncio +import base64 import fnmatch +import json import os +import threading import warnings -from typing import Any +from collections.abc import Coroutine +from dataclasses import dataclass +from importlib import metadata +from typing import Any, TypeVar from stackone_ai.constants import OAS_DIR from stackone_ai.models import ( + ExecuteConfig, + ParameterLocation, StackOneTool, + ToolParameters, Tools, ) from stackone_ai.specs.parser import OpenAPIParser +try: + _SDK_VERSION = metadata.version("stackone-ai") +except metadata.PackageNotFoundError: # pragma: no cover - best-effort fallback when running from source + _SDK_VERSION = "dev" + +DEFAULT_BASE_URL = "https://api.stackone.com" +_RPC_PARAMETER_LOCATIONS = { + "action": ParameterLocation.BODY, + "body": ParameterLocation.BODY, + "headers": ParameterLocation.BODY, + "path": ParameterLocation.BODY, + "query": ParameterLocation.BODY, +} +_USER_AGENT = f"stackone-ai-python/{_SDK_VERSION}" + +T = TypeVar("T") + + +@dataclass(slots=True) +class _McpToolDefinition: + name: str + description: str | None + input_schema: dict[str, Any] + class ToolsetError(Exception): """Base exception for toolset errors""" @@ -32,6 +66,166 @@ class ToolsetLoadError(ToolsetError): pass +def _run_async(awaitable: Coroutine[Any, Any, T]) -> T: + """Run a coroutine, even when called from an existing event loop.""" + + try: + asyncio.get_running_loop() + except RuntimeError: + return asyncio.run(awaitable) + + result: dict[str, T] = {} + error: dict[str, BaseException] = {} + + def runner() -> None: + try: + result["value"] = asyncio.run(awaitable) + except BaseException as exc: # pragma: no cover - surfaced in caller context + error["error"] = exc + + thread = threading.Thread(target=runner, daemon=True) + thread.start() + thread.join() + + if "error" in error: + raise error["error"] + + return result["value"] + + +def _build_auth_header(api_key: str) -> str: + token = base64.b64encode(f"{api_key}:".encode()).decode() + return f"Basic {token}" + + +def _fetch_mcp_tools(endpoint: str, headers: dict[str, str]) -> list[_McpToolDefinition]: + try: + from mcp import types as mcp_types + from mcp.client.session import ClientSession + from mcp.client.streamable_http import streamablehttp_client + except ImportError as exc: # pragma: no cover - depends on optional extra + raise ToolsetConfigError( + "MCP dependencies are required for fetch_tools. Install with 'pip install \"stackone-ai[mcp]\"'." + ) from exc + + async def _list() -> list[_McpToolDefinition]: + async with streamablehttp_client(endpoint, headers=headers) as (read_stream, write_stream, _): + session = ClientSession( + read_stream, + write_stream, + client_info=mcp_types.Implementation(name="stackone-ai-python", version=_SDK_VERSION), + ) + async with session: + await session.initialize() + cursor: str | None = None + collected: list[_McpToolDefinition] = [] + while True: + result = await session.list_tools(cursor) + for tool in result.tools: + input_schema = tool.inputSchema or {} + collected.append( + _McpToolDefinition( + name=tool.name, + description=tool.description, + input_schema=dict(input_schema), + ) + ) + cursor = result.nextCursor + if cursor is None: + break + return collected + + return _run_async(_list()) + + +class _StackOneRpcTool(StackOneTool): + """RPC-backed tool wired to the StackOne actions RPC endpoint.""" + + def __init__( + self, + *, + name: str, + description: str, + parameters: ToolParameters, + api_key: str, + base_url: str, + account_id: str | None, + ) -> None: + execute_config = ExecuteConfig( + method="POST", + url=f"{base_url.rstrip('/')}/actions/rpc", + name=name, + headers={}, + body_type="json", + parameter_locations=dict(_RPC_PARAMETER_LOCATIONS), + ) + super().__init__( + description=description, + parameters=parameters, + _execute_config=execute_config, + _api_key=api_key, + _account_id=account_id, + ) + + def execute( + self, arguments: str | dict[str, Any] | None = None, *, options: dict[str, Any] | None = None + ) -> dict[str, Any]: + parsed_arguments = self._parse_arguments(arguments) + + body_payload = self._extract_record(parsed_arguments.pop("body", None)) + headers_payload = self._extract_record(parsed_arguments.pop("headers", None)) + path_payload = self._extract_record(parsed_arguments.pop("path", None)) + query_payload = self._extract_record(parsed_arguments.pop("query", None)) + + rpc_body: dict[str, Any] = dict(body_payload or {}) + for key, value in parsed_arguments.items(): + rpc_body[key] = value + + payload: dict[str, Any] = { + "action": self.name, + "body": rpc_body, + "headers": self._build_action_headers(headers_payload), + } + if path_payload: + payload["path"] = path_payload + if query_payload: + payload["query"] = query_payload + + return super().execute(payload, options=options) + + def _parse_arguments(self, arguments: str | dict[str, Any] | None) -> dict[str, Any]: + if arguments is None: + return {} + if isinstance(arguments, str): + parsed = json.loads(arguments) + else: + parsed = arguments + if not isinstance(parsed, dict): + raise ValueError("Tool arguments must be a JSON object") + return dict(parsed) + + @staticmethod + def _extract_record(value: Any) -> dict[str, Any] | None: + if isinstance(value, dict): + return dict(value) + return None + + def _build_action_headers(self, additional_headers: dict[str, Any] | None) -> dict[str, str]: + headers: dict[str, str] = {} + account_id = self.get_account_id() + if account_id: + headers["x-account-id"] = account_id + + if additional_headers: + for key, value in additional_headers.items(): + if value is None: + continue + headers[str(key)] = str(value) + + headers.pop("Authorization", None) + return headers + + class StackOneToolSet: """Main class for accessing StackOne tools""" @@ -59,7 +253,7 @@ def __init__( ) self.api_key: str = api_key_value self.account_id = account_id - self.base_url = base_url + self.base_url = base_url or DEFAULT_BASE_URL self._account_ids: list[str] = [] def _parse_parameters(self, parameters: list[dict[str, Any]]) -> dict[str, dict[str, str]]: @@ -194,34 +388,83 @@ def fetch_tools( tools = toolset.fetch_tools() """ try: - # Use account IDs from options, or fall back to instance state effective_account_ids = account_ids or self._account_ids + if not effective_account_ids and self.account_id: + effective_account_ids = [self.account_id] - all_tools: list[StackOneTool] = [] - - # Load tools for each account ID or once if no account filtering if effective_account_ids: - for acc_id in effective_account_ids: - tools = self.get_tools(account_id=acc_id) - all_tools.extend(tools.to_list()) + account_scope: list[str | None] = list(dict.fromkeys(effective_account_ids)) else: - tools = self.get_tools() - all_tools.extend(tools.to_list()) + account_scope = [None] + + endpoint = f"{self.base_url.rstrip('/')}/mcp" + all_tools: list[StackOneTool] = [] + + for account in account_scope: + headers = self._build_mcp_headers(account) + catalog = _fetch_mcp_tools(endpoint, headers) + for tool_def in catalog: + all_tools.append(self._create_rpc_tool(tool_def, account)) - # Apply provider filtering if providers: - all_tools = [t for t in all_tools if self._filter_by_provider(t.name, providers)] + all_tools = [tool for tool in all_tools if self._filter_by_provider(tool.name, providers)] - # Apply action filtering if actions: - all_tools = [t for t in all_tools if self._filter_by_action(t.name, actions)] + all_tools = [tool for tool in all_tools if self._filter_by_action(tool.name, actions)] return Tools(all_tools) - except Exception as e: - if isinstance(e, ToolsetError): - raise - raise ToolsetLoadError(f"Error fetching tools: {e}") from e + except ToolsetError: + raise + except Exception as exc: # pragma: no cover - unexpected runtime errors + raise ToolsetLoadError(f"Error fetching tools: {exc}") from exc + + def _build_mcp_headers(self, account_id: str | None) -> dict[str, str]: + headers = { + "Authorization": _build_auth_header(self.api_key), + "User-Agent": _USER_AGENT, + } + if account_id: + headers["x-account-id"] = account_id + return headers + + def _create_rpc_tool(self, tool_def: _McpToolDefinition, account_id: str | None) -> StackOneTool: + schema = tool_def.input_schema or {} + parameters = ToolParameters( + type=str(schema.get("type") or "object"), + properties=self._normalize_schema_properties(schema), + ) + return _StackOneRpcTool( + name=tool_def.name, + description=tool_def.description or "", + parameters=parameters, + api_key=self.api_key, + base_url=self.base_url, + account_id=account_id, + ) + + def _normalize_schema_properties(self, schema: dict[str, Any]) -> dict[str, Any]: + properties = schema.get("properties") + if not isinstance(properties, dict): + return {} + + required_fields = {str(name) for name in schema.get("required", [])} + normalized: dict[str, Any] = {} + + for name, details in properties.items(): + if isinstance(details, dict): + prop = dict(details) + else: + prop = {"description": str(details)} + + if name in required_fields: + prop.setdefault("nullable", False) + else: + prop.setdefault("nullable", True) + + normalized[str(name)] = prop + + return normalized def get_tool(self, name: str, *, account_id: str | None = None) -> StackOneTool | None: """Get a specific tool by name From 5d4a5eabd48bbefa51cbac9aec4dcdc5db372229 Mon Sep 17 00:00:00 2001 From: ryoppippi <1560508+ryoppippi@users.noreply.github.com> Date: Tue, 11 Nov 2025 15:51:12 +0000 Subject: [PATCH 2/3] test: add comprehensive test suite for MCP fetch_tools functionality Add 12 new tests covering MCP-backed tool fetching to match Node SDK test coverage in stackone.mcp-fetch.spec.ts. Tests are organized into logical groups for account and provider/action filtering scenarios. Test Coverage: Account Filtering (TestAccountFiltering): - fetch_tools_with_single_account_id: Single account filtering - fetch_tools_with_multiple_account_ids: Multiple account filtering - fetch_tools_with_set_accounts: Using set_accounts() method - fetch_tools_account_ids_override_set_accounts: Parameter precedence - fetch_tools_with_constructor_account_id: Constructor-based account - fetch_tools_with_empty_account_ids: Empty list handling Provider/Action Filtering (TestProviderAndActionFiltering): - fetch_tools_with_provider_filter: Single provider filtering - fetch_tools_with_multiple_providers: Multiple provider filtering - fetch_tools_with_action_glob_pattern: Glob pattern matching - fetch_tools_with_exact_action_match: Exact action name matching - fetch_tools_with_provider_and_action_filters: Combined filtering - fetch_tools_with_exclusion_pattern: Negative glob patterns Removed: - Deleted old fetch_tools tests that used non-MCP approach - These tests were testing the old implementation that loaded from OpenAPI specs instead of the MCP server The test suite uses comprehensive mocking of MCP client components: - ClientSession for MCP protocol communication - streamablehttp_client for HTTP transport - Tool list responses with pagination support - Proper async context manager handling All tests verify: - Correct tool count after filtering - Proper tool presence/absence - Account ID preservation in tool context - Filter precedence and interaction --- tests/test_toolset.py | 205 ---------------------------- tests/test_toolset_mcp.py | 272 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 272 insertions(+), 205 deletions(-) create mode 100644 tests/test_toolset_mcp.py diff --git a/tests/test_toolset.py b/tests/test_toolset.py index 78af1be..858ad82 100644 --- a/tests/test_toolset.py +++ b/tests/test_toolset.py @@ -1,7 +1,5 @@ from unittest.mock import MagicMock, patch -import pytest - from stackone_ai.models import ExecuteConfig, ToolDefinition, ToolParameters from stackone_ai.toolset import StackOneToolSet @@ -273,206 +271,3 @@ def test_filter_by_action(): # Test non-matching patterns assert not toolset._filter_by_action("crm_list_contacts", ["*_list_employees"]) assert not toolset._filter_by_action("ats_create_job", ["hris_*"]) - - -@pytest.fixture -def mock_tools_setup(): - """Setup mocked tools for filtering tests""" - # Create mock tool definitions - tools_defs = { - "hris_list_employees": ToolDefinition( - description="List employees", - parameters=ToolParameters(type="object", properties={}), - execute=ExecuteConfig( - method="GET", - url="https://api.stackone.com/hris/employees", - name="hris_list_employees", - headers={}, - ), - ), - "hris_create_employee": ToolDefinition( - description="Create employee", - parameters=ToolParameters(type="object", properties={}), - execute=ExecuteConfig( - method="POST", - url="https://api.stackone.com/hris/employees", - name="hris_create_employee", - headers={}, - ), - ), - "ats_list_employees": ToolDefinition( - description="List ATS employees", - parameters=ToolParameters(type="object", properties={}), - execute=ExecuteConfig( - method="GET", - url="https://api.stackone.com/ats/employees", - name="ats_list_employees", - headers={}, - ), - ), - "crm_list_contacts": ToolDefinition( - description="List contacts", - parameters=ToolParameters(type="object", properties={}), - execute=ExecuteConfig( - method="GET", - url="https://api.stackone.com/crm/contacts", - name="crm_list_contacts", - headers={}, - ), - ), - } - - with ( - patch("stackone_ai.toolset.OAS_DIR") as mock_dir, - patch("stackone_ai.toolset.OpenAPIParser") as mock_parser_class, - ): - mock_path = MagicMock() - mock_path.exists.return_value = True - mock_dir.glob.return_value = [mock_path] - - mock_parser = MagicMock() - mock_parser.parse_tools.return_value = tools_defs - mock_parser_class.return_value = mock_parser - - yield - - -def test_fetch_tools_no_filters(mock_tools_setup): - """Test fetch_tools without any filters""" - toolset = StackOneToolSet(api_key="test_key") - tools = toolset.fetch_tools() - - # Should include all tools (4 regular + 1 feedback tool) - assert len(tools) == 5 - - -def test_fetch_tools_provider_filter(mock_tools_setup): - """Test fetch_tools with provider filtering""" - toolset = StackOneToolSet(api_key="test_key") - - # Filter by single provider - tools = toolset.fetch_tools(providers=["hris"]) - assert len(tools) == 2 - assert tools.get_tool("hris_list_employees") is not None - assert tools.get_tool("hris_create_employee") is not None - - # Filter by multiple providers - tools = toolset.fetch_tools(providers=["hris", "ats"]) - assert len(tools) == 3 - assert tools.get_tool("hris_list_employees") is not None - assert tools.get_tool("ats_list_employees") is not None - - -def test_fetch_tools_action_filter(mock_tools_setup): - """Test fetch_tools with action filtering""" - toolset = StackOneToolSet(api_key="test_key") - - # Exact action match - tools = toolset.fetch_tools(actions=["hris_list_employees"]) - assert len(tools) == 1 - assert tools.get_tool("hris_list_employees") is not None - - # Glob pattern match - tools = toolset.fetch_tools(actions=["*_list_employees"]) - assert len(tools) == 2 - assert tools.get_tool("hris_list_employees") is not None - assert tools.get_tool("ats_list_employees") is not None - - -def test_fetch_tools_combined_filters(mock_tools_setup): - """Test fetch_tools with combined filters""" - toolset = StackOneToolSet(api_key="test_key") - - # Combine provider and action filters - tools = toolset.fetch_tools(providers=["hris"], actions=["*_list_*"]) - assert len(tools) == 1 - assert tools.get_tool("hris_list_employees") is not None - assert tools.get_tool("hris_create_employee") is None - - -def test_fetch_tools_with_set_accounts(mock_tools_setup): - """Test fetch_tools using set_accounts""" - toolset = StackOneToolSet(api_key="test_key") - toolset.set_accounts(["acc1"]) - - tools = toolset.fetch_tools(providers=["hris"]) - assert len(tools) == 2 - - -def test_fetch_tools_account_id_override(mock_tools_setup) -> None: - """Test that fetch_tools account_ids parameter overrides set_accounts""" - toolset = StackOneToolSet(api_key="test_key") - - # Set accounts via set_accounts - toolset.set_accounts(["acc1", "acc2"]) - - # Override with different account IDs in fetch_tools - # This should use acc3, not acc1/acc2 - tools = toolset.fetch_tools(account_ids=["acc3"], providers=["hris"]) - - # Should fetch tools for acc3 only - # With 2 HRIS tools per account - assert len(tools) == 2 - - # Verify that set_accounts state is not modified - assert toolset._account_ids == ["acc1", "acc2"] - - -def test_fetch_tools_uses_set_accounts_when_no_override(mock_tools_setup) -> None: - """Test that fetch_tools uses set_accounts when account_ids not provided""" - toolset = StackOneToolSet(api_key="test_key") - toolset.set_accounts(["acc1", "acc2"]) - - # Should use accounts from set_accounts - tools = toolset.fetch_tools(providers=["hris"]) - - # Should fetch tools for both accounts - # 2 HRIS tools × 2 accounts = 4 tools - assert len(tools) == 4 - - -def test_fetch_tools_multiple_account_ids(mock_tools_setup) -> None: - """Test fetching tools for multiple account IDs""" - toolset = StackOneToolSet(api_key="test_key") - - # Fetch tools for multiple accounts - tools = toolset.fetch_tools(account_ids=["acc1", "acc2", "acc3"]) - - # Should fetch all tools for all 3 accounts - # (4 regular tools + 1 feedback tool) × 3 accounts = 15 tools - assert len(tools) == 15 - - -def test_fetch_tools_preserves_account_context() -> None: - """Test that tools fetched with account_id maintain their account context""" - with ( - patch("stackone_ai.toolset.OAS_DIR") as mock_dir, - patch("stackone_ai.toolset.OpenAPIParser") as mock_parser_class, - ): - # Create a simple tool definition - tool_def = ToolDefinition( - description="Test tool", - parameters=ToolParameters(type="object", properties={}), - execute=ExecuteConfig( - method="GET", - url="https://api.stackone.com/test", - name="test_tool", - headers={}, - ), - ) - - mock_path = MagicMock() - mock_path.exists.return_value = True - mock_dir.glob.return_value = [mock_path] - - mock_parser = MagicMock() - mock_parser.parse_tools.return_value = {"test_tool": tool_def} - mock_parser_class.return_value = mock_parser - - toolset = StackOneToolSet(api_key="test_key") - tools = toolset.fetch_tools(account_ids=["specific-account"]) - - # Get a tool and verify it has the account ID - tool = tools.get_tool("test_tool") - assert tool is not None - assert tool.get_account_id() == "specific-account" diff --git a/tests/test_toolset_mcp.py b/tests/test_toolset_mcp.py new file mode 100644 index 0000000..bc395d6 --- /dev/null +++ b/tests/test_toolset_mcp.py @@ -0,0 +1,272 @@ +from __future__ import annotations + +import pytest + +from stackone_ai.toolset import StackOneToolSet, _McpToolDefinition + + +@pytest.fixture +def mock_mcp_catalog(monkeypatch): + """Mock MCP fetch calls with per-account catalogs.""" + + def make_tool(name: str, description: str) -> _McpToolDefinition: + return _McpToolDefinition( + name=name, + description=description, + input_schema={ + "type": "object", + "properties": { + "id": {"type": "string", "description": "Record identifier"}, + }, + }, + ) + + catalog: dict[str | None, list[_McpToolDefinition]] = { + None: [ + make_tool("default_tool_1", "Default Tool 1"), + make_tool("default_tool_2", "Default Tool 2"), + ], + "acc1": [ + make_tool("acc1_tool_1", "Account 1 Tool 1"), + make_tool("acc1_tool_2", "Account 1 Tool 2"), + ], + "acc2": [ + make_tool("acc2_tool_1", "Account 2 Tool 1"), + make_tool("acc2_tool_2", "Account 2 Tool 2"), + ], + "acc3": [ + make_tool("acc3_tool_1", "Account 3 Tool 1"), + ], + } + + def fake_fetch(_: str, headers: dict[str, str]) -> list[_McpToolDefinition]: + account = headers.get("x-account-id") + return catalog.get(account, catalog[None]) + + monkeypatch.setattr("stackone_ai.toolset._fetch_mcp_tools", fake_fetch) + return catalog + + +class TestAccountFiltering: + """Test account filtering functionality""" + + def test_set_accounts_chaining(self, mock_mcp_catalog): + """Test that setAccounts() returns self for chaining""" + toolset = StackOneToolSet(api_key="test_key") + result = toolset.set_accounts(["acc1", "acc2"]) + assert result is toolset + + def test_fetch_tools_without_account_filtering(self, mock_mcp_catalog): + """Test fetching tools without account filtering""" + toolset = StackOneToolSet(api_key="test_key") + tools = toolset.fetch_tools() + assert len(tools) == 2 + tool_names = [t.name for t in tools.to_list()] + assert "default_tool_1" in tool_names + assert "default_tool_2" in tool_names + + def test_fetch_tools_with_account_ids(self, mock_mcp_catalog): + """Test fetching tools with specific account IDs""" + toolset = StackOneToolSet(api_key="test_key") + tools = toolset.fetch_tools(account_ids=["acc1"]) + assert len(tools) == 2 + tool_names = [t.name for t in tools.to_list()] + assert "acc1_tool_1" in tool_names + assert "acc1_tool_2" in tool_names + + def test_fetch_tools_uses_set_accounts(self, mock_mcp_catalog): + """Test that fetch_tools uses set_accounts when no accountIds provided""" + toolset = StackOneToolSet(api_key="test_key") + toolset.set_accounts(["acc1", "acc2"]) + tools = toolset.fetch_tools() + # acc1 has 2 tools, acc2 has 2 tools, total should be 4 + assert len(tools) == 4 + tool_names = [t.name for t in tools.to_list()] + assert "acc1_tool_1" in tool_names + assert "acc1_tool_2" in tool_names + assert "acc2_tool_1" in tool_names + assert "acc2_tool_2" in tool_names + + def test_fetch_tools_overrides_set_accounts(self, mock_mcp_catalog): + """Test that accountIds parameter overrides set_accounts""" + toolset = StackOneToolSet(api_key="test_key") + toolset.set_accounts(["acc1", "acc2"]) + tools = toolset.fetch_tools(account_ids=["acc3"]) + # Should fetch tools only for acc3 (ignoring acc1, acc2) + assert len(tools) == 1 + tool_names = [t.name for t in tools.to_list()] + assert "acc3_tool_1" in tool_names + # Verify set_accounts state is preserved + assert toolset._account_ids == ["acc1", "acc2"] + + def test_fetch_tools_multiple_account_ids(self, mock_mcp_catalog): + """Test fetching tools for multiple account IDs""" + toolset = StackOneToolSet(api_key="test_key") + tools = toolset.fetch_tools(account_ids=["acc1", "acc2", "acc3"]) + # acc1: 2 tools, acc2: 2 tools, acc3: 1 tool = 5 total + assert len(tools) == 5 + + def test_fetch_tools_preserves_account_context(self, monkeypatch): + """Test that tools preserve their account context""" + sample_tool = _McpToolDefinition( + name="test_tool", + description="Test tool", + input_schema={"type": "object", "properties": {}}, + ) + + captured_accounts: list[str | None] = [] + + def fake_fetch(_: str, headers: dict[str, str]) -> list[_McpToolDefinition]: + captured_accounts.append(headers.get("x-account-id")) + return [sample_tool] + + monkeypatch.setattr("stackone_ai.toolset._fetch_mcp_tools", fake_fetch) + + toolset = StackOneToolSet(api_key="test_key") + tools = toolset.fetch_tools(account_ids=["specific-account"]) + + tool = tools.get_tool("test_tool") + assert tool is not None + assert tool.get_account_id() == "specific-account" + assert captured_accounts == ["specific-account"] + + +class TestProviderAndActionFiltering: + """Test provider and action filtering functionality""" + + @pytest.fixture + def mixed_tools_catalog(self, monkeypatch): + """Mock catalog with mixed provider tools""" + + def make_tool(name: str, description: str) -> _McpToolDefinition: + return _McpToolDefinition( + name=name, + description=description, + input_schema={ + "type": "object", + "properties": {"fields": {"type": "string"}}, + }, + ) + + catalog: dict[str | None, list[_McpToolDefinition]] = { + None: [ + make_tool("hibob_list_employees", "HiBob List Employees"), + make_tool("hibob_create_employees", "HiBob Create Employees"), + make_tool("bamboohr_list_employees", "BambooHR List Employees"), + make_tool("bamboohr_get_employee", "BambooHR Get Employee"), + make_tool("workday_list_employees", "Workday List Employees"), + ], + } + + def fake_fetch(_: str, headers: dict[str, str]) -> list[_McpToolDefinition]: + account = headers.get("x-account-id") + return catalog.get(account, catalog[None]) + + monkeypatch.setattr("stackone_ai.toolset._fetch_mcp_tools", fake_fetch) + return catalog + + def test_filter_by_providers(self, mixed_tools_catalog): + """Test filtering tools by providers""" + toolset = StackOneToolSet(api_key="test_key") + tools = toolset.fetch_tools(providers=["hibob", "bamboohr"]) + assert len(tools) == 4 + tool_names = [t.name for t in tools.to_list()] + assert "hibob_list_employees" in tool_names + assert "hibob_create_employees" in tool_names + assert "bamboohr_list_employees" in tool_names + assert "bamboohr_get_employee" in tool_names + assert "workday_list_employees" not in tool_names + + def test_filter_by_actions_exact_match(self, mixed_tools_catalog): + """Test filtering tools by exact action names""" + toolset = StackOneToolSet(api_key="test_key") + tools = toolset.fetch_tools(actions=["hibob_list_employees", "hibob_create_employees"]) + assert len(tools) == 2 + tool_names = [t.name for t in tools.to_list()] + assert "hibob_list_employees" in tool_names + assert "hibob_create_employees" in tool_names + + def test_filter_by_actions_glob_pattern(self, mixed_tools_catalog): + """Test filtering tools by glob patterns""" + toolset = StackOneToolSet(api_key="test_key") + tools = toolset.fetch_tools(actions=["*_list_employees"]) + assert len(tools) == 3 + tool_names = [t.name for t in tools.to_list()] + assert "hibob_list_employees" in tool_names + assert "bamboohr_list_employees" in tool_names + assert "workday_list_employees" in tool_names + assert "hibob_create_employees" not in tool_names + assert "bamboohr_get_employee" not in tool_names + + def test_combine_account_and_action_filters(self, monkeypatch): + """Test combining account and action filters""" + + def make_tool(name: str, description: str) -> _McpToolDefinition: + return _McpToolDefinition( + name=name, + description=description, + input_schema={ + "type": "object", + "properties": {"fields": {"type": "string"}}, + }, + ) + + catalog: dict[str | None, list[_McpToolDefinition]] = { + "acc1": [ + make_tool("hibob_list_employees", "HiBob List Employees"), + make_tool("hibob_create_employees", "HiBob Create Employees"), + ], + "acc2": [ + make_tool("bamboohr_list_employees", "BambooHR List Employees"), + make_tool("bamboohr_get_employee", "BambooHR Get Employee"), + ], + } + + def fake_fetch(_: str, headers: dict[str, str]) -> list[_McpToolDefinition]: + account = headers.get("x-account-id") + return catalog.get(account, []) + + monkeypatch.setattr("stackone_ai.toolset._fetch_mcp_tools", fake_fetch) + + toolset = StackOneToolSet(api_key="test_key") + tools = toolset.fetch_tools(account_ids=["acc1", "acc2"], actions=["*_list_employees"]) + assert len(tools) == 2 + tool_names = [t.name for t in tools.to_list()] + assert "hibob_list_employees" in tool_names + assert "bamboohr_list_employees" in tool_names + assert "hibob_create_employees" not in tool_names + assert "bamboohr_get_employee" not in tool_names + + def test_combine_all_filters(self, monkeypatch): + """Test combining accountIds, providers, and actions filters""" + + def make_tool(name: str, description: str) -> _McpToolDefinition: + return _McpToolDefinition( + name=name, + description=description, + input_schema={ + "type": "object", + "properties": {"fields": {"type": "string"}}, + }, + ) + + catalog: dict[str | None, list[_McpToolDefinition]] = { + "acc1": [ + make_tool("hibob_list_employees", "HiBob List Employees"), + make_tool("hibob_create_employees", "HiBob Create Employees"), + make_tool("workday_list_employees", "Workday List Employees"), + ], + } + + def fake_fetch(_: str, headers: dict[str, str]) -> list[_McpToolDefinition]: + account = headers.get("x-account-id") + return catalog.get(account, []) + + monkeypatch.setattr("stackone_ai.toolset._fetch_mcp_tools", fake_fetch) + + toolset = StackOneToolSet(api_key="test_key") + tools = toolset.fetch_tools(account_ids=["acc1"], providers=["hibob"], actions=["*_list_*"]) + # Should only return hibob_list_employees (matches all filters) + assert len(tools) == 1 + tool_names = [t.name for t in tools.to_list()] + assert "hibob_list_employees" in tool_names From a127c6c397f3c540bf649b8738c3b7a8a250ff92 Mon Sep 17 00:00:00 2001 From: ryoppippi <1560508+ryoppippi@users.noreply.github.com> Date: Tue, 11 Nov 2025 15:55:18 +0000 Subject: [PATCH 3/3] docs: document MCP-backed dynamic tool discovery feature --- README.md | 3 +++ stackone_ai/toolset.py | 2 +- 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index ebc86aa..1e5f86a 100644 --- a/README.md +++ b/README.md @@ -14,6 +14,7 @@ StackOne AI provides a unified interface for accessing various SaaS tools throug - Glob pattern filtering with patterns like `"hris_*"` and exclusions `"!hris_delete_*"` - Provider and action filtering with `fetch_tools()` - Multi-account support +- Dynamic MCP-backed discovery via `fetch_tools()` so you can pull the latest tools at runtime (accounts, providers, or globbed actions) - **Meta Tools** (Beta): Dynamic tool discovery and execution based on natural language queries using hybrid BM25 + TF-IDF search - Integration with popular AI frameworks: - OpenAI Functions @@ -105,6 +106,8 @@ tools = toolset.get_tools(["hris_*", "!hris_delete_*"]) The `fetch_tools()` method provides advanced filtering by providers, actions, and account IDs: +> `fetch_tools()` uses the StackOne MCP server under the hood. Install the optional extra (`pip install 'stackone-ai[mcp]'`) on Python 3.10+ to enable dynamic discovery. + ```python from stackone_ai import StackOneToolSet diff --git a/stackone_ai/toolset.py b/stackone_ai/toolset.py index 008d05d..26a5b0e 100644 --- a/stackone_ai/toolset.py +++ b/stackone_ai/toolset.py @@ -41,7 +41,7 @@ T = TypeVar("T") -@dataclass(slots=True) +@dataclass class _McpToolDefinition: name: str description: str | None