Coverage for stackone_ai / feedback / tool.py: 96%
79 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-02-08 18:25 +0000
1"""Feedback collection tool for StackOne."""
3from __future__ import annotations
5import json
7from pydantic import BaseModel, Field, field_validator
9from ..models import (
10 ExecuteConfig,
11 JsonDict,
12 ParameterLocation,
13 StackOneError,
14 StackOneTool,
15 ToolParameters,
16)
19class FeedbackInput(BaseModel):
20 """Input schema for feedback tool."""
22 feedback: str = Field(..., min_length=1, description="User feedback text")
23 account_id: str | list[str] = Field(..., description="Account identifier(s) - single ID or list of IDs")
24 tool_names: list[str] = Field(..., min_length=1, description="List of tool names")
26 @field_validator("feedback")
27 @classmethod
28 def validate_feedback(cls, v: str) -> str:
29 """Validate that feedback is non-empty after trimming."""
30 trimmed = v.strip()
31 if not trimmed:
32 raise ValueError("Feedback must be a non-empty string")
33 return trimmed
35 @field_validator("account_id")
36 @classmethod
37 def validate_account_id(cls, v: str | list[str]) -> list[str]:
38 """Validate and normalize account ID(s) to a list."""
39 if isinstance(v, str):
40 trimmed = v.strip()
41 if not trimmed:
42 raise ValueError("Account ID must be a non-empty string")
43 return [trimmed]
45 if isinstance(v, list): 45 ↛ 53line 45 didn't jump to line 53 because the condition on line 45 was always true
46 if not v:
47 raise ValueError("At least one account ID is required")
48 cleaned = [str(item).strip() for item in v if str(item).strip()]
49 if not cleaned:
50 raise ValueError("At least one valid account ID is required")
51 return cleaned
53 raise ValueError("Account ID must be a string or list of strings")
55 @field_validator("tool_names")
56 @classmethod
57 def validate_tool_names(cls, v: list[str]) -> list[str]:
58 """Validate and clean tool names."""
59 cleaned = [name.strip() for name in v if name.strip()]
60 if not cleaned:
61 raise ValueError("At least one tool name is required")
62 return cleaned
65class FeedbackTool(StackOneTool):
66 """Extended tool for collecting feedback with enhanced validation."""
68 def execute(
69 self, arguments: str | JsonDict | None = None, *, options: JsonDict | None = None
70 ) -> JsonDict:
71 """
72 Execute the feedback tool with enhanced validation.
74 If multiple account IDs are provided, sends the same feedback to each account individually.
76 Args:
77 arguments: Tool arguments as string or dict
78 options: Execution options
80 Returns:
81 Combined response from all API calls
83 Raises:
84 StackOneError: If validation or API call fails
85 """
86 try:
87 # Parse input
88 if isinstance(arguments, str):
89 raw_params = json.loads(arguments)
90 else:
91 raw_params = arguments or {}
93 # Validate with Pydantic
94 parsed_params = FeedbackInput(**raw_params)
96 # Get list of account IDs (already normalized by validator)
97 account_ids = parsed_params.account_id
98 feedback = parsed_params.feedback
99 tool_names = parsed_params.tool_names
101 # If only one account ID, use the parent execute method
102 if len(account_ids) == 1:
103 validated_arguments = {
104 "feedback": feedback,
105 "account_id": account_ids[0],
106 "tool_names": tool_names,
107 }
108 return super().execute(validated_arguments, options=options)
110 # Multiple account IDs - send to each individually
111 results = []
112 errors = []
114 for account_id in account_ids:
115 try:
116 validated_arguments = {
117 "feedback": feedback,
118 "account_id": account_id,
119 "tool_names": tool_names,
120 }
121 result = super().execute(validated_arguments, options=options)
122 results.append({"account_id": account_id, "status": "success", "result": result})
123 except Exception as exc:
124 error_msg = str(exc)
125 errors.append({"account_id": account_id, "status": "error", "error": error_msg})
126 results.append({"account_id": account_id, "status": "error", "error": error_msg})
128 # Return combined results
129 return {
130 "message": f"Feedback sent to {len(account_ids)} account(s)",
131 "total_accounts": len(account_ids),
132 "successful": len([r for r in results if r["status"] == "success"]),
133 "failed": len(errors),
134 "results": results,
135 }
137 except json.JSONDecodeError as exc:
138 raise StackOneError(f"Invalid JSON in arguments: {exc}") from exc
139 except ValueError as exc:
140 raise StackOneError(f"Validation error: {exc}") from exc
141 except Exception as error:
142 if isinstance(error, StackOneError): 142 ↛ 144line 142 didn't jump to line 144 because the condition on line 142 was always true
143 raise
144 raise StackOneError(f"Error executing feedback tool: {error}") from error
147def create_feedback_tool(
148 api_key: str,
149 account_id: str | None = None,
150 base_url: str = "https://api.stackone.com",
151) -> FeedbackTool:
152 """
153 Create a feedback collection tool.
155 Args:
156 api_key: API key for authentication
157 account_id: Optional account ID
158 base_url: Base URL for the API
160 Returns:
161 FeedbackTool configured for feedback collection
162 """
163 name = "tool_feedback"
164 description = (
165 "Collects user feedback on StackOne tool performance. "
166 'First ask the user, "Are you ok with sending feedback to StackOne?" '
167 "and mention that the LLM will take care of sending it. "
168 "Call this tool only when the user explicitly answers yes."
169 )
171 parameters = ToolParameters(
172 type="object",
173 properties={
174 "account_id": {
175 "oneOf": [
176 {
177 "type": "string",
178 "description": 'Single account identifier (e.g., "acc_123456")',
179 },
180 {
181 "type": "array",
182 "items": {"type": "string"},
183 "description": "List of account identifiers for multiple accounts",
184 },
185 ],
186 "description": "Account identifier(s) - single ID or list of IDs",
187 },
188 "feedback": {
189 "type": "string",
190 "description": "Verbatim feedback from the user about their experience with StackOne tools.",
191 },
192 "tool_names": {
193 "type": "array",
194 "items": {
195 "type": "string",
196 },
197 "description": "Array of tool names being reviewed",
198 },
199 },
200 )
202 execute_config = ExecuteConfig(
203 name=name,
204 method="POST",
205 url=f"{base_url}/ai/tool-feedback",
206 body_type="json",
207 parameter_locations={
208 "feedback": ParameterLocation.BODY,
209 "account_id": ParameterLocation.BODY,
210 "tool_names": ParameterLocation.BODY,
211 },
212 )
214 # Create instance by calling parent class __init__ directly since FeedbackTool is a subclass
215 tool = FeedbackTool.__new__(FeedbackTool)
216 StackOneTool.__init__(
217 tool,
218 description=description,
219 parameters=parameters,
220 _execute_config=execute_config,
221 _api_key=api_key,
222 _account_id=account_id,
223 )
225 return tool