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 e5b2235..26a5b0e 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 +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 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