[Ready][toolkit/zendesk] Adding WhoAmI tool to zendesk (#544)

Zendesk who_am_i tool.

---------

Co-authored-by: Francisco Liberal <francisco@arcade.dev>

toolkits/zendesk/arcade_zendesk/tools/__init__.py

@@ -1,4 +1,5 @@
 from arcade_zendesk.tools.search_articles import search_articles
+from arcade_zendesk.tools.system_context import who_am_i
 from arcade_zendesk.tools.tickets import (
     add_ticket_comment,
     get_ticket_comments,
@@ -12,4 +13,5 @@ __all__ = [
     "get_ticket_comments",
     "mark_ticket_solved",
     "search_articles",
+    "who_am_i",
 ]

toolkits/zendesk/arcade_zendesk/tools/system_context.py

@@ -0,0 +1,26 @@
+from typing import Annotated, Any
+
+from arcade_tdk import ToolContext, tool
+from arcade_tdk.auth import OAuth2
+
+from arcade_zendesk.who_am_i_util import build_who_am_i_response
+
+
+@tool(
+    requires_auth=OAuth2(id="zendesk", scopes=["read"]),
+    requires_secrets=["ZENDESK_SUBDOMAIN"],
+)
+async def who_am_i(
+    context: ToolContext,
+) -> Annotated[
+    dict[str, Any],
+    "Get comprehensive user profile and Zendesk account information.",
+]:
+    """
+    Get comprehensive user profile and Zendesk account information.
+
+    This tool provides detailed information about the authenticated user including
+    their name, email, role, organization details, and Zendesk account context.
+    """
+    user_info = await build_who_am_i_response(context)
+    return dict(user_info)

toolkits/zendesk/arcade_zendesk/who_am_i_util.py

@@ -0,0 +1,120 @@
+from typing import Any, TypedDict
+
+import httpx
+from arcade_tdk import ToolContext
+
+
+class WhoAmIResponse(TypedDict, total=False):
+    user_id: int
+    name: str
+    email: str
+    role: str
+    active: bool
+    verified: bool
+    locale: str
+    time_zone: str
+    organization_id: int
+    organization_name: str
+    organization_domains: list[str]
+    zendesk_access: bool
+
+
+async def build_who_am_i_response(context: ToolContext) -> WhoAmIResponse:
+    """Build comprehensive who am I response for Zendesk."""
+    user_info = await _get_current_user(context)
+    organization_info = await _get_organization_info(context, user_info.get("organization_id"))
+
+    response_data = {}
+    response_data.update(_extract_user_info(user_info))
+    response_data.update(_extract_organization_info(organization_info))
+    response_data["zendesk_access"] = True
+
+    return response_data  # type: ignore[return-value]
+
+
+async def _get_current_user(context: ToolContext) -> dict[str, Any]:
+    """Get current user information from Zendesk API."""
+    subdomain = context.get_secret("ZENDESK_SUBDOMAIN")
+    base_url = f"https://{subdomain}.zendesk.com"
+
+    headers = {
+        "Authorization": f"Bearer {context.get_auth_token_or_empty()}",
+        "Content-Type": "application/json",
+    }
+
+    async with httpx.AsyncClient() as client:
+        response = await client.get(f"{base_url}/api/v2/users/me", headers=headers)
+        response.raise_for_status()
+        return response.json().get("user", {})  # type: ignore[no-any-return]
+
+
+async def _get_organization_info(
+    context: ToolContext, organization_id: int | None
+) -> dict[str, Any]:
+    """Get organization information from Zendesk API."""
+    if not organization_id:
+        return {}
+
+    subdomain = context.get_secret("ZENDESK_SUBDOMAIN")
+    base_url = f"https://{subdomain}.zendesk.com"
+
+    headers = {
+        "Authorization": f"Bearer {context.get_auth_token_or_empty()}",
+        "Content-Type": "application/json",
+    }
+
+    async with httpx.AsyncClient() as client:
+        response = await client.get(
+            f"{base_url}/api/v2/organizations/{organization_id}", headers=headers
+        )
+        response.raise_for_status()
+        return response.json().get("organization", {})  # type: ignore[no-any-return]
+
+
+def _extract_user_info(user_info: dict[str, Any]) -> dict[str, Any]:
+    """Extract user information from Zendesk user response."""
+    extracted = {}
+
+    if user_info.get("id"):
+        extracted["user_id"] = user_info["id"]
+
+    if user_info.get("name"):
+        extracted["name"] = user_info["name"]
+
+    if user_info.get("email"):
+        extracted["email"] = user_info["email"]
+
+    if user_info.get("role"):
+        extracted["role"] = user_info["role"]
+
+    if "active" in user_info:
+        extracted["active"] = user_info["active"]
+
+    if "verified" in user_info:
+        extracted["verified"] = user_info["verified"]
+
+    if user_info.get("locale"):
+        extracted["locale"] = user_info["locale"]
+
+    if user_info.get("time_zone"):
+        extracted["time_zone"] = user_info["time_zone"]
+
+    if user_info.get("organization_id"):
+        extracted["organization_id"] = user_info["organization_id"]
+
+    return extracted
+
+
+def _extract_organization_info(organization_info: dict[str, Any]) -> dict[str, Any]:
+    """Extract organization information from Zendesk organization response."""
+    extracted = {}
+
+    if organization_info.get("name"):
+        extracted["organization_name"] = organization_info["name"]
+
+    if organization_info.get("domain_names"):
+        domains = organization_info["domain_names"]
+        if domains:
+            extracted["organization_domains"] = domains
+
+    return extracted

toolkits/zendesk/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "arcade_zendesk"
-version = "0.1.0"
+version = "0.2.0"
 requires-python = ">=3.10"
 dependencies = [
     "arcade-tdk>=2.0.0,<3.0.0",

toolkits/zendesk/tests/test_who_am_i_util.py

@@ -0,0 +1,330 @@
+from unittest.mock import Mock, patch
+
+import httpx
+import pytest
+
+from arcade_zendesk.who_am_i_util import (
+    WhoAmIResponse,
+    _extract_organization_info,
+    _extract_user_info,
+    _get_current_user,
+    _get_organization_info,
+    build_who_am_i_response,
+)
+
+
+@pytest.fixture
+def mock_context():
+    """Create a mock ToolContext for testing."""
+    context = Mock()
+    context.get_secret.return_value = "test-subdomain"
+    context.get_auth_token_or_empty.return_value = "test-token"
+    return context
+
+
+@pytest.fixture
+def sample_user_data():
+    """Sample user data from Zendesk API."""
+    return {
+        "id": 12345,
+        "name": "John Doe",
+        "email": "john.doe@example.com",
+        "role": "admin",
+        "active": True,
+        "verified": True,
+        "locale": "en-US",
+        "time_zone": "America/New_York",
+        "organization_id": 67890,
+        "created_at": "2023-01-01T00:00:00Z",
+        "updated_at": "2023-12-01T00:00:00Z",
+    }
+
+
+@pytest.fixture
+def sample_organization_data():
+    """Sample organization data from Zendesk API."""
+    return {
+        "id": 67890,
+        "name": "Example Corp",
+        "domain_names": ["example.com", "example.org"],
+        "created_at": "2022-01-01T00:00:00Z",
+        "updated_at": "2023-11-01T00:00:00Z",
+        "details": "Main organization",
+        "notes": "Primary customer",
+        "group_id": 123,
+        "shared_tickets": True,
+        "shared_comments": False,
+    }
+
+
+class TestBuildWhoAmIResponse:
+    """Test the main build_who_am_i_response function."""
+
+    @pytest.mark.asyncio
+    async def test_build_complete_response(
+        self, mock_context, sample_user_data, sample_organization_data
+    ):
+        """Test building a complete who am I response."""
+        with (
+            patch("arcade_zendesk.who_am_i_util._get_current_user") as mock_get_user,
+            patch("arcade_zendesk.who_am_i_util._get_organization_info") as mock_get_org,
+        ):
+            mock_get_user.return_value = sample_user_data
+            mock_get_org.return_value = sample_organization_data
+
+            result = await build_who_am_i_response(mock_context)
+
+            assert isinstance(result, dict)
+            assert result["user_id"] == 12345
+            assert result["name"] == "John Doe"
+            assert result["email"] == "john.doe@example.com"
+            assert result["role"] == "admin"
+            assert result["active"] is True
+            assert result["verified"] is True
+            assert result["locale"] == "en-US"
+            assert result["time_zone"] == "America/New_York"
+            assert result["organization_id"] == 67890
+            assert result["organization_name"] == "Example Corp"
+            assert result["organization_domains"] == ["example.com", "example.org"]
+            assert result["zendesk_access"] is True
+
+            mock_get_user.assert_called_once_with(mock_context)
+            mock_get_org.assert_called_once_with(mock_context, 67890)
+
+    @pytest.mark.asyncio
+    async def test_build_response_without_organization(self, mock_context, sample_user_data):
+        """Test building response when user has no organization."""
+        user_data_no_org = sample_user_data.copy()
+        del user_data_no_org["organization_id"]
+
+        with (
+            patch("arcade_zendesk.who_am_i_util._get_current_user") as mock_get_user,
+            patch("arcade_zendesk.who_am_i_util._get_organization_info") as mock_get_org,
+        ):
+            mock_get_user.return_value = user_data_no_org
+            mock_get_org.return_value = {}
+
+            result = await build_who_am_i_response(mock_context)
+
+            assert result["user_id"] == 12345
+            assert result["name"] == "John Doe"
+            assert result["zendesk_access"] is True
+            assert "organization_name" not in result
+            assert "organization_domains" not in result
+
+            mock_get_org.assert_called_once_with(mock_context, None)
+
+
+class TestGetCurrentUser:
+    """Test the _get_current_user function."""
+
+    @pytest.mark.asyncio
+    async def test_get_current_user_success(self, mock_context, sample_user_data):
+        """Test successful user retrieval."""
+        mock_response = Mock()
+        mock_response.json.return_value = {"user": sample_user_data}
+        mock_response.raise_for_status.return_value = None
+
+        with patch("httpx.AsyncClient") as mock_client:
+            mock_client.return_value.__aenter__.return_value.get.return_value = mock_response
+
+            result = await _get_current_user(mock_context)
+
+            assert result == sample_user_data
+            mock_client.return_value.__aenter__.return_value.get.assert_called_once_with(
+                "https://test-subdomain.zendesk.com/api/v2/users/me",
+                headers={
+                    "Authorization": "Bearer test-token",
+                    "Content-Type": "application/json",
+                },
+            )
+
+    @pytest.mark.asyncio
+    async def test_get_current_user_http_error(self, mock_context):
+        """Test user retrieval with HTTP error."""
+        with patch("httpx.AsyncClient") as mock_client:
+            mock_client.return_value.__aenter__.return_value.get.side_effect = (
+                httpx.HTTPStatusError("404 Not Found", request=Mock(), response=Mock())
+            )
+
+            with pytest.raises(httpx.HTTPStatusError):
+                await _get_current_user(mock_context)
+
+
+class TestGetOrganizationInfo:
+    """Test the _get_organization_info function."""
+
+    @pytest.mark.asyncio
+    async def test_get_organization_info_success(self, mock_context, sample_organization_data):
+        """Test successful organization retrieval."""
+        mock_response = Mock()
+        mock_response.json.return_value = {"organization": sample_organization_data}
+        mock_response.raise_for_status.return_value = None
+
+        with patch("httpx.AsyncClient") as mock_client:
+            mock_client.return_value.__aenter__.return_value.get.return_value = mock_response
+
+            result = await _get_organization_info(mock_context, 67890)
+
+            assert result == sample_organization_data
+            mock_client.return_value.__aenter__.return_value.get.assert_called_once_with(
+                "https://test-subdomain.zendesk.com/api/v2/organizations/67890",
+                headers={
+                    "Authorization": "Bearer test-token",
+                    "Content-Type": "application/json",
+                },
+            )
+
+    @pytest.mark.asyncio
+    async def test_get_organization_info_no_id(self, mock_context):
+        """Test organization retrieval with no organization ID."""
+        result = await _get_organization_info(mock_context, None)
+        assert result == {}
+
+    @pytest.mark.asyncio
+    async def test_get_organization_info_http_error(self, mock_context):
+        """Test organization retrieval with HTTP error."""
+        with patch("httpx.AsyncClient") as mock_client:
+            mock_client.return_value.__aenter__.return_value.get.side_effect = (
+                httpx.HTTPStatusError("404 Not Found", request=Mock(), response=Mock())
+            )
+
+            with pytest.raises(httpx.HTTPStatusError):
+                await _get_organization_info(mock_context, 67890)
+
+
+class TestExtractUserInfo:
+    """Test the _extract_user_info function."""
+
+    def test_extract_complete_user_info(self, sample_user_data):
+        """Test extracting complete user information."""
+        result = _extract_user_info(sample_user_data)
+
+        expected = {
+            "user_id": 12345,
+            "name": "John Doe",
+            "email": "john.doe@example.com",
+            "role": "admin",
+            "active": True,
+            "verified": True,
+            "locale": "en-US",
+            "time_zone": "America/New_York",
+            "organization_id": 67890,
+        }
+
+        assert result == expected
+
+    def test_extract_partial_user_info(self):
+        """Test extracting partial user information."""
+        partial_data = {
+            "id": 12345,
+            "name": "John Doe",
+            "email": "john.doe@example.com",
+        }
+
+        result = _extract_user_info(partial_data)
+
+        expected = {
+            "user_id": 12345,
+            "name": "John Doe",
+            "email": "john.doe@example.com",
+        }
+
+        assert result == expected
+
+    def test_extract_empty_user_info(self):
+        """Test extracting from empty user data."""
+        result = _extract_user_info({})
+        assert result == {}
+
+    @pytest.mark.parametrize(
+        "field,value,expected_key",
+        [
+            ("active", False, "active"),
+            ("verified", False, "verified"),
+            ("active", True, "active"),
+            ("verified", True, "verified"),
+        ],
+    )
+    def test_extract_boolean_fields(self, field, value, expected_key):
+        """Test extracting boolean fields correctly."""
+        user_data = {field: value}
+        result = _extract_user_info(user_data)
+        assert result[expected_key] == value
+
+
+class TestExtractOrganizationInfo:
+    """Test the _extract_organization_info function."""
+
+    def test_extract_complete_organization_info(self, sample_organization_data):
+        """Test extracting complete organization information."""
+        result = _extract_organization_info(sample_organization_data)
+
+        assert result["organization_name"] == "Example Corp"
+        assert result["organization_domains"] == ["example.com", "example.org"]
+
+    def test_extract_organization_info_no_domains(self):
+        """Test extracting organization info without domain names."""
+        org_data = {
+            "name": "Example Corp",
+            "domain_names": [],
+        }
+
+        result = _extract_organization_info(org_data)
+
+        assert result["organization_name"] == "Example Corp"
+        assert "organization_domains" not in result
+
+    def test_extract_organization_info_multiple_domains(self):
+        """Test extracting organization info with multiple domains."""
+        org_data = {
+            "name": "Example Corp",
+            "domain_names": ["primary.com", "secondary.com", "tertiary.com"],
+        }
+
+        result = _extract_organization_info(org_data)
+
+        assert result["organization_name"] == "Example Corp"
+        assert result["organization_domains"] == ["primary.com", "secondary.com", "tertiary.com"]
+
+    def test_extract_empty_organization_info(self):
+        """Test extracting from empty organization data."""
+        result = _extract_organization_info({})
+        assert "organization_name" not in result
+        assert "organization_domains" not in result
+        assert result == {}
+
+
+class TestWhoAmIResponseType:
+    """Test the WhoAmIResponse TypedDict."""
+
+    def test_typed_dict_structure(self):
+        """Test that WhoAmIResponse accepts expected fields."""
+        response: WhoAmIResponse = {
+            "user_id": 12345,
+            "name": "John Doe",
+            "email": "john.doe@example.com",
+            "role": "admin",
+            "active": True,
+            "verified": True,
+            "locale": "en-US",
+            "time_zone": "America/New_York",
+            "organization_id": 67890,
+            "organization_name": "Example Corp",
+            "organization_domains": ["example.com", "example.org"],
+            "zendesk_access": True,
+        }
+
+        assert response["user_id"] == 12345
+        assert response["zendesk_access"] is True
+
+    def test_typed_dict_partial(self):
+        """Test that WhoAmIResponse works with partial data."""
+        response: WhoAmIResponse = {
+            "user_id": 12345,
+            "name": "John Doe",
+            "zendesk_access": True,
+        }
+
+        assert response["user_id"] == 12345
+        assert response["zendesk_access"] is True