From 1e1ad5d7dd8af591bdc6e7a776f19067b0cb558d Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 6 Feb 2026 11:02:57 +0000 Subject: [PATCH] feat(tools): add Deadlock API schema fetch tool and enhance API info - Enrich DEADLOCK_API_INFO with overview, docs links, and detailed descriptions for both Game Data and Assets APIs - Add DeadlockAPISchemaFetchTool that fetches OpenAPI specs for either the data API or assets API so the agent can answer detailed questions about available endpoints, parameters, and response formats - Register the new tool in ToolRegistry with proper lifecycle management - Update system prompt with Deadlock API section including all resource links and guidance to use the schema tool for API usage questions - Add comprehensive tests for both the info tool and schema fetch tool https://claude.ai/code/session_017pU3sCe7ku3xp7as1VXnkp --- packages/ai_assistant/agent.py | 14 ++ packages/api/patreon_test.py | 4 +- packages/tools/openapi/deadlock_api.py | 147 ++++++++++++++- packages/tools/openapi/openapi_test.py | 248 +++++++++++++++++++++++++ packages/tools/registry.py | 22 +++ 5 files changed, 422 insertions(+), 13 deletions(-) diff --git a/packages/ai_assistant/agent.py b/packages/ai_assistant/agent.py index 9fff876..eb4d533 100644 --- a/packages/ai_assistant/agent.py +++ b/packages/ai_assistant/agent.py @@ -241,6 +241,20 @@ async def tool_permission_handler( **3. Hero & Meta** * **Stats**: Weapon, Vitality, Spirit. **Flex Slots**: Unlocked by destroying objectives. * **Beta Status**: Game is in closed-beta. Patches are frequent; prioritize tool data over memory. + +**4. Deadlock API** +The Deadlock API is a community-driven, open-source project providing comprehensive game data access. +* **Main Website**: https://deadlock-api.com/ +* **Game Data API**: https://api.deadlock-api.com/ — matches, players, leaderboards, hero/item stats, MMR. + * Swagger Docs: https://api.deadlock-api.com/docs +* **Assets API**: https://assets.deadlock-api.com/ — hero portraits, item icons, ability details, game metadata. + * Swagger Docs: https://assets.deadlock-api.com/docs +* **GitHub**: https://github.com/deadlock-api/ +* **Discord**: https://discord.gg/XMF9Xrgfqu +* **Patreon**: https://www.patreon.com/user?u=68961896 +* When users ask about available API endpoints, how to use the API, or what data the API provides, + use the `deadlock_api_schema` tool to fetch the full OpenAPI spec and provide accurate answers. +* Use `deadlock_api_info` for quick resource links and general API information. diff --git a/packages/api/patreon_test.py b/packages/api/patreon_test.py index 3fde28f..1188b86 100644 --- a/packages/api/patreon_test.py +++ b/packages/api/patreon_test.py @@ -538,9 +538,7 @@ async def test_fetch_user_identity_success_with_active_patron(self) -> None: }, "relationships": { "campaign": {"data": {"id": "test-campaign-id"}}, - "currently_entitled_tiers": { - "data": [{"type": "tier", "id": tier.tier_id}] - }, + "currently_entitled_tiers": {"data": [{"type": "tier", "id": tier.tier_id}]}, }, } ], diff --git a/packages/tools/openapi/deadlock_api.py b/packages/tools/openapi/deadlock_api.py index 3276f45..8774419 100644 --- a/packages/tools/openapi/deadlock_api.py +++ b/packages/tools/openapi/deadlock_api.py @@ -73,34 +73,64 @@ async def create_deadlock_api_tools( # Deadlock API resource URLs DEADLOCK_API_INFO = { + "overview": { + "description": ( + "The Deadlock API is a community-driven, open-source project providing comprehensive data access " + "for Valve's game Deadlock. It offers two main APIs: a Game Data API for match data, player stats, " + "leaderboards, and analytics; and an Assets API for hero portraits, item icons, ability images, " + "and other game media. Both APIs are free to use, well-documented via OpenAPI/Swagger specs, and " + "do not require authentication for most endpoints." + ), + }, "main_website": { "url": "https://deadlock-api.com/", - "description": "Main website for the Deadlock API project", - }, - "assets_api": { - "url": "https://assets.deadlock-api.com/", - "description": "API for game assets including images, sounds, hero portraits, item icons, and media files", - "openapi_spec": "https://assets.deadlock-api.com/openapi.json", + "description": ( + "Main website for the Deadlock API project with overview, documentation links, and getting started guides" + ), }, "data_api": { "url": "https://api.deadlock-api.com/", - "description": "API for match data, player statistics, analytics, match histories, leaderboards, and more", + "description": ( + "Game Data API — provides match data, player statistics, hero win rates, item pick rates, " + "leaderboards, match histories, MMR tracking, and more. Endpoints include match details, " + "player profiles, hero/item stats, and ranked leaderboards." + ), "openapi_spec": "https://api.deadlock-api.com/openapi.json", + "docs": "https://api.deadlock-api.com/docs", + }, + "assets_api": { + "url": "https://assets.deadlock-api.com/", + "description": ( + "Assets API — provides game assets including hero portraits, ability icons, item icons, " + "hero stats, item stats, ability details, and raw game data. Use this to get hero/item metadata, " + "images, and detailed game balance information." + ), + "openapi_spec": "https://assets.deadlock-api.com/openapi.json", + "docs": "https://assets.deadlock-api.com/docs", }, "github": { "url": "https://github.com/deadlock-api/", - "description": "GitHub organization with source code, documentation, and community contributions", + "description": ( + "GitHub organization with source code, documentation, and community contributions " + "for all Deadlock API projects" + ), }, "discord": { "url": "https://discord.gg/XMF9Xrgfqu", - "description": "Discord community server for support, discussions, and announcements", + "description": "Discord community server for API support, bug reports, feature requests, and announcements", }, "patreon": { "url": "https://www.patreon.com/user?u=68961896", - "description": "Patreon page to support the Deadlock API project development", + "description": "Patreon page to support ongoing Deadlock API development and server costs", }, } +# OpenAPI spec URLs for schema fetching +OPENAPI_SPEC_URLS: dict[str, str] = { + "data": "https://api.deadlock-api.com/openapi.json", + "assets": "https://assets.deadlock-api.com/openapi.json", +} + class DeadlockAPIInfoTool(BaseTool): """Tool that returns information about the Deadlock API resources. @@ -151,6 +181,101 @@ def _create_result_summary(self, result: dict[str, Any]) -> str: return f"Returned info for {len(result)} Deadlock API resources" +class DeadlockAPISchemaFetchTool(BaseTool): + """Tool that fetches OpenAPI schemas for the Deadlock APIs. + + Allows the agent to retrieve the full OpenAPI specification for either + the Game Data API or the Assets API, enabling it to answer detailed + questions about available endpoints, parameters, and response formats. + """ + + def __init__( + self, + sse_callback: SSECallback, + timeout: float = 60.0, + ) -> None: + super().__init__(sse_callback, timeout) + self._http_client: httpx.AsyncClient | None = None + + @property + def name(self) -> str: + return "deadlock_api_schema" + + async def _get_client(self) -> httpx.AsyncClient: + """Get or create the HTTP client.""" + if self._http_client is None: + self._http_client = httpx.AsyncClient(timeout=self._timeout) + return self._http_client + + def get_definition(self) -> dict[str, Any]: + """Get tool definition for agent configuration.""" + return { + "name": self.name, + "description": ( + "Fetch the OpenAPI schema for a Deadlock API. Use this to answer questions about " + "available API endpoints, request parameters, response formats, and how to use the API. " + "Pass api='data' for the Game Data API (matches, players, leaderboards) or " + "api='assets' for the Assets API (heroes, items, images, game data)." + ), + "parameters": { + "type": "object", + "properties": { + "api": { + "type": "string", + "enum": ["data", "assets"], + "description": ( + "Which API schema to fetch: " + "'data' for the Game Data API (api.deadlock-api.com) or " + "'assets' for the Assets API (assets.deadlock-api.com)" + ), + }, + }, + "required": ["api"], + }, + } + + @retry(max_attempts=3, base_delay=1.0) + async def _run(self, api: str) -> dict[str, Any]: + """Fetch the OpenAPI schema for the specified API. + + Args: + api: Which API to fetch the schema for ('data' or 'assets') + + Returns: + The OpenAPI specification as a dictionary + + Raises: + ValueError: If api is not 'data' or 'assets' + OpenAPIConnectionError: If the schema cannot be fetched + """ + if api not in OPENAPI_SPEC_URLS: + raise ValueError(f"Invalid API name: {api}. Must be one of: {', '.join(sorted(OPENAPI_SPEC_URLS))}") + + spec_url = OPENAPI_SPEC_URLS[api] + + try: + client = await self._get_client() + response = await client.get(spec_url) + response.raise_for_status() + return response.json() + except httpx.HTTPStatusError as e: + raise OpenAPIConnectionError(f"Failed to fetch {api} API schema: HTTP {e.response.status_code}") from e + except httpx.RequestError as e: + raise OpenAPIConnectionError(f"Network error fetching {api} API schema: {e}") from e + + def _create_result_summary(self, result: dict[str, Any]) -> str: + info = result.get("info", {}) + title = info.get("title", "Unknown API") + paths = result.get("paths", {}) + return f"Schema for {title}: {len(paths)} endpoints" + + async def close(self) -> None: + """Close the HTTP client.""" + if self._http_client: + await self._http_client.aclose() + self._http_client = None + + class DeadlockAPICallTool(BaseTool): """Generic tool for calling any Deadlock API endpoint. @@ -283,11 +408,13 @@ async def close(self) -> None: "DeadlockAPIToolGenerator", "DeadlockAPICallTool", "DeadlockAPIInfoTool", + "DeadlockAPISchemaFetchTool", "create_deadlock_api_tools", "DEADLOCK_API_SPEC_URL", "DEADLOCK_API_TOOL_PREFIX", "DEADLOCK_API_BASE_URL", "DEADLOCK_API_EXCLUDED_OPERATIONS", "DEADLOCK_API_INFO", + "OPENAPI_SPEC_URLS", "VALID_HTTP_METHODS", ] diff --git a/packages/tools/openapi/openapi_test.py b/packages/tools/openapi/openapi_test.py index 873ed69..f580e46 100644 --- a/packages/tools/openapi/openapi_test.py +++ b/packages/tools/openapi/openapi_test.py @@ -19,8 +19,12 @@ ) from packages.tools.openapi.deadlock_api import ( DEADLOCK_API_BASE_URL, + DEADLOCK_API_INFO, + OPENAPI_SPEC_URLS, VALID_HTTP_METHODS, DeadlockAPICallTool, + DeadlockAPIInfoTool, + DeadlockAPISchemaFetchTool, DeadlockAPIToolGenerator, ) @@ -1364,6 +1368,250 @@ def test_result_summary_formatting(self, item_mapping_tool: GetItemMappingTool) assert "Item mapping: 2 items" in summary +class TestDeadlockAPIInfoTool: + """Tests for DeadlockAPIInfoTool.""" + + @pytest.fixture + def sse_callback(self) -> Any: + """Create SSE callback.""" + + def callback(event: ChatToolStartEvent | ChatToolEndEvent) -> None: + pass # noqa: ARG001 + + return callback + + @pytest.fixture + def info_tool(self, sse_callback: Any) -> DeadlockAPIInfoTool: + """Create DeadlockAPIInfoTool instance.""" + return DeadlockAPIInfoTool(sse_callback=sse_callback, timeout=10.0) + + def test_name_property(self, info_tool: DeadlockAPIInfoTool) -> None: + """Test name property returns correct name.""" + assert info_tool.name == "deadlock_api_info" + + def test_get_definition(self, info_tool: DeadlockAPIInfoTool) -> None: + """Test get_definition returns correct structure.""" + definition = info_tool.get_definition() + assert definition["name"] == "deadlock_api_info" + assert "Deadlock API" in definition["description"] + + @pytest.mark.asyncio + async def test_run_returns_api_info(self, info_tool: DeadlockAPIInfoTool) -> None: + """Test _run returns the DEADLOCK_API_INFO dictionary.""" + result = await info_tool._run() + assert result == DEADLOCK_API_INFO + + def test_api_info_contains_overview(self) -> None: + """Test DEADLOCK_API_INFO includes an overview section.""" + assert "overview" in DEADLOCK_API_INFO + assert "description" in DEADLOCK_API_INFO["overview"] + + def test_api_info_contains_data_api_with_docs(self) -> None: + """Test DEADLOCK_API_INFO includes data API with docs link.""" + assert "data_api" in DEADLOCK_API_INFO + data_api = DEADLOCK_API_INFO["data_api"] + assert "url" in data_api + assert "openapi_spec" in data_api + assert "docs" in data_api + + def test_api_info_contains_assets_api_with_docs(self) -> None: + """Test DEADLOCK_API_INFO includes assets API with docs link.""" + assert "assets_api" in DEADLOCK_API_INFO + assets_api = DEADLOCK_API_INFO["assets_api"] + assert "url" in assets_api + assert "openapi_spec" in assets_api + assert "docs" in assets_api + + def test_api_info_contains_community_links(self) -> None: + """Test DEADLOCK_API_INFO includes GitHub, Discord, and Patreon links.""" + assert "github" in DEADLOCK_API_INFO + assert "discord" in DEADLOCK_API_INFO + assert "patreon" in DEADLOCK_API_INFO + + def test_result_summary(self, info_tool: DeadlockAPIInfoTool) -> None: + """Test result summary formatting.""" + summary = info_tool._create_result_summary(DEADLOCK_API_INFO) + assert f"{len(DEADLOCK_API_INFO)}" in summary + + +class TestDeadlockAPISchemaFetchTool: + """Tests for DeadlockAPISchemaFetchTool.""" + + @pytest.fixture + def sse_events(self) -> list[ChatToolStartEvent | ChatToolEndEvent]: + """Capture SSE events.""" + return [] + + @pytest.fixture + def sse_callback(self, sse_events: list[ChatToolStartEvent | ChatToolEndEvent]) -> Any: + """Create SSE callback that captures events.""" + + def callback(event: ChatToolStartEvent | ChatToolEndEvent) -> None: + sse_events.append(event) + + return callback + + @pytest.fixture + def schema_tool(self, sse_callback: Any) -> DeadlockAPISchemaFetchTool: + """Create DeadlockAPISchemaFetchTool instance.""" + return DeadlockAPISchemaFetchTool(sse_callback=sse_callback, timeout=10.0) + + def test_name_property(self, schema_tool: DeadlockAPISchemaFetchTool) -> None: + """Test name property returns correct name.""" + assert schema_tool.name == "deadlock_api_schema" + + def test_get_definition(self, schema_tool: DeadlockAPISchemaFetchTool) -> None: + """Test get_definition returns correct structure.""" + definition = schema_tool.get_definition() + assert definition["name"] == "deadlock_api_schema" + assert "api" in definition["parameters"]["properties"] + assert "api" in definition["parameters"]["required"] + assert definition["parameters"]["properties"]["api"]["enum"] == ["data", "assets"] + + @pytest.mark.asyncio + async def test_run_fetches_data_api_schema(self, schema_tool: DeadlockAPISchemaFetchTool) -> None: + """Test fetching the data API schema.""" + sample_spec = { + "openapi": "3.0.0", + "info": {"title": "Deadlock API", "version": "1.0.0"}, + "paths": {"/v1/matches": {"get": {"summary": "Get matches"}}}, + } + + mock_response = MagicMock() + mock_response.json.return_value = sample_spec + mock_response.raise_for_status = MagicMock() + + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=mock_response) + + with patch.object(schema_tool, "_get_client", return_value=mock_client): + result = await schema_tool._run(api="data") + + assert result == sample_spec + mock_client.get.assert_called_once_with(OPENAPI_SPEC_URLS["data"]) + + @pytest.mark.asyncio + async def test_run_fetches_assets_api_schema(self, schema_tool: DeadlockAPISchemaFetchTool) -> None: + """Test fetching the assets API schema.""" + sample_spec = { + "openapi": "3.0.0", + "info": {"title": "Deadlock Assets API", "version": "1.0.0"}, + "paths": {"/v2/heroes": {"get": {"summary": "Get heroes"}}}, + } + + mock_response = MagicMock() + mock_response.json.return_value = sample_spec + mock_response.raise_for_status = MagicMock() + + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=mock_response) + + with patch.object(schema_tool, "_get_client", return_value=mock_client): + result = await schema_tool._run(api="assets") + + assert result == sample_spec + mock_client.get.assert_called_once_with(OPENAPI_SPEC_URLS["assets"]) + + @pytest.mark.asyncio + async def test_run_rejects_invalid_api_name(self, schema_tool: DeadlockAPISchemaFetchTool) -> None: + """Test ValueError raised for invalid API name.""" + with pytest.raises(ValueError) as exc_info: + await schema_tool._run(api="invalid") + + assert "Invalid API name" in str(exc_info.value) + + @pytest.mark.asyncio + async def test_run_handles_http_error(self, schema_tool: DeadlockAPISchemaFetchTool) -> None: + """Test handling of HTTP errors during schema fetch.""" + mock_response = MagicMock() + mock_response.status_code = 503 + error = httpx.HTTPStatusError("Service Unavailable", request=MagicMock(), response=mock_response) + + mock_client = AsyncMock() + mock_client.get = AsyncMock(side_effect=error) + + with ( + patch.object(schema_tool, "_get_client", return_value=mock_client), + pytest.raises(OpenAPIConnectionError) as exc_info, + ): + await schema_tool._run(api="data") + + assert "HTTP 503" in str(exc_info.value) + + @pytest.mark.asyncio + async def test_run_handles_network_error(self, schema_tool: DeadlockAPISchemaFetchTool) -> None: + """Test handling of network errors during schema fetch.""" + error = httpx.RequestError("Connection refused") + + mock_client = AsyncMock() + mock_client.get = AsyncMock(side_effect=error) + + with ( + patch.object(schema_tool, "_get_client", return_value=mock_client), + pytest.raises(OpenAPIConnectionError) as exc_info, + ): + await schema_tool._run(api="assets") + + assert "Network error" in str(exc_info.value) + + @pytest.mark.asyncio + async def test_execute_emits_sse_events( + self, schema_tool: DeadlockAPISchemaFetchTool, sse_events: list[ChatToolStartEvent | ChatToolEndEvent] + ) -> None: + """Test execute emits start and end SSE events.""" + sample_spec = { + "openapi": "3.0.0", + "info": {"title": "Deadlock API", "version": "1.0.0"}, + "paths": {"/v1/matches": {}, "/v1/players": {}}, + } + + mock_response = MagicMock() + mock_response.json.return_value = sample_spec + mock_response.raise_for_status = MagicMock() + + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=mock_response) + + with patch.object(schema_tool, "_get_client", return_value=mock_client): + await schema_tool.execute(api="data") + + assert len(sse_events) == 2 + + start_event = sse_events[0] + assert isinstance(start_event, ChatToolStartEvent) + assert start_event.tool_name == "deadlock_api_schema" + assert start_event.arguments == {"api": "data"} + + end_event = sse_events[1] + assert isinstance(end_event, ChatToolEndEvent) + assert end_event.success is True + assert "2 endpoints" in end_event.result_summary + + def test_result_summary_formatting(self, schema_tool: DeadlockAPISchemaFetchTool) -> None: + """Test result summary formatting.""" + result = { + "info": {"title": "Deadlock API"}, + "paths": {"/v1/matches": {}, "/v1/players": {}, "/v1/heroes": {}}, + } + summary = schema_tool._create_result_summary(result) + assert "Deadlock API" in summary + assert "3 endpoints" in summary + + def test_result_summary_with_missing_info(self, schema_tool: DeadlockAPISchemaFetchTool) -> None: + """Test result summary with missing info section.""" + result = {"paths": {"/v1/matches": {}}} + summary = schema_tool._create_result_summary(result) + assert "Unknown API" in summary + assert "1 endpoints" in summary + + def test_openapi_spec_urls_constant(self) -> None: + """Test OPENAPI_SPEC_URLS contains expected entries.""" + assert "data" in OPENAPI_SPEC_URLS + assert "assets" in OPENAPI_SPEC_URLS + assert OPENAPI_SPEC_URLS["data"] == "https://api.deadlock-api.com/openapi.json" + assert OPENAPI_SPEC_URLS["assets"] == "https://assets.deadlock-api.com/openapi.json" + + class TestDeadlockAPIToolGenerator: """Tests for DeadlockAPIToolGenerator.""" diff --git a/packages/tools/registry.py b/packages/tools/registry.py index 79501b8..1eece65 100644 --- a/packages/tools/registry.py +++ b/packages/tools/registry.py @@ -18,6 +18,7 @@ from packages.tools.openapi.deadlock_api import ( DeadlockAPICallTool, DeadlockAPIInfoTool, + DeadlockAPISchemaFetchTool, DeadlockAPIToolGenerator, ) from packages.tools.wiki import WikiGetPageTool, WikiSearchTool @@ -61,6 +62,7 @@ def __init__(self, sse_callback: SSECallback, timeout: float = 60.0) -> None: self._deadlock_api_tools: dict[str, OpenAPITool] | None = None self._deadlock_api_call_tool: DeadlockAPICallTool | None = None self._deadlock_api_info_tool: DeadlockAPIInfoTool | None = None + self._deadlock_api_schema_tool: DeadlockAPISchemaFetchTool | None = None self._assets_api_generator: AssetsAPIToolGenerator | None = None self._assets_api_tools: dict[str, OpenAPITool] | None = None self._helper_tools: dict[str, BaseTool] | None = None @@ -117,6 +119,15 @@ def _get_deadlock_api_info_tool(self) -> DeadlockAPIInfoTool: ) return self._deadlock_api_info_tool + def _get_deadlock_api_schema_tool(self) -> DeadlockAPISchemaFetchTool: + """Lazy initialization of Deadlock API schema fetch tool.""" + if self._deadlock_api_schema_tool is None: + self._deadlock_api_schema_tool = DeadlockAPISchemaFetchTool( + sse_callback=self._sse_callback, + timeout=self._timeout, + ) + return self._deadlock_api_schema_tool + def _get_assets_api_generator(self) -> AssetsAPIToolGenerator: """Lazy initialization of Assets API generator.""" if self._assets_api_generator is None: @@ -210,6 +221,13 @@ async def get_all_tools(self) -> dict[str, BaseTool | OpenAPITool]: except Exception as e: self._warnings.append(ToolLoadWarning("Deadlock API Info", str(e), e)) + # Deadlock API schema fetch tool + try: + deadlock_api_schema = self._get_deadlock_api_schema_tool() + self._all_tools[deadlock_api_schema.name] = deadlock_api_schema + except Exception as e: + self._warnings.append(ToolLoadWarning("Deadlock API Schema", str(e), e)) + # Assets API tools (dynamic from OpenAPI spec) try: assets_api_tools = await self._get_assets_api_tools() @@ -296,6 +314,10 @@ async def close(self) -> None: if self._deadlock_api_call_tool is not None: await self._deadlock_api_call_tool.close() + # Close Deadlock API schema fetch tool + if self._deadlock_api_schema_tool is not None: + await self._deadlock_api_schema_tool.close() + # Close Assets API generator if self._assets_api_generator is not None: await self._assets_api_generator.close()