refactor: address Pwuts review feedback

- Remove unused error_response() and format_inputs_as_markdown() from helpers.py
- Remove _get_inputs_list() wrapper from run_agent.py, use get_inputs_from_schema directly
- Fix type annotations: get_user_credentials, find_matching_credential, create_credential_meta_from_match
- Remove check_scopes parameter - always check scopes (original missing check was broken behavior)
- Reorder _credential_has_required_scopes to be defined before find_matching_credential
This commit is contained in:
Otto
2026-02-03 16:57:38 +00:00
parent 10ebaf5104
commit 435738f2f9
2 changed files with 3 additions and 59 deletions

View File

@@ -2,39 +2,12 @@
from typing import Any
from .models import ErrorResponse
def error_response(
message: str, session_id: str | None, **kwargs: Any
) -> ErrorResponse:
"""Create standardized error response.
Args:
message: Error message to display
session_id: Current session ID
**kwargs: Additional fields to pass to ErrorResponse
Returns:
ErrorResponse with the given message and session_id
"""
return ErrorResponse(message=message, session_id=session_id, **kwargs)
def get_inputs_from_schema(
input_schema: dict[str, Any],
exclude_fields: set[str] | None = None,
) -> list[dict[str, Any]]:
"""Extract input field info from JSON schema.
Args:
input_schema: JSON schema dict with 'properties' and 'required'
exclude_fields: Set of field names to exclude (e.g., credential fields)
Returns:
List of dicts with field info (name, title, type, description, required, default)
"""
# Safety check: original code returned [] if input_schema wasn't a dict
"""Extract input field info from JSON schema."""
if not isinstance(input_schema, dict):
return []
@@ -54,28 +27,3 @@ def get_inputs_from_schema(
for name, schema in properties.items()
if name not in exclude
]
def format_inputs_as_markdown(inputs: list[dict[str, Any]]) -> str:
"""Format input fields as a readable markdown list.
Args:
inputs: List of input dicts from get_inputs_from_schema
Returns:
Markdown-formatted string listing the inputs
"""
if not inputs:
return "No inputs required."
lines = []
for inp in inputs:
required_marker = " (required)" if inp.get("required") else ""
default = inp.get("default")
default_info = f" [default: {default}]" if default is not None else ""
description = inp.get("description", "")
desc_info = f" - {description}" if description else ""
lines.append(f"- **{inp['name']}**{required_marker}{default_info}{desc_info}")
return "\n".join(lines)

View File

@@ -262,7 +262,7 @@ class RunAgentTool(BaseTool):
),
requirements={
"credentials": requirements_creds_list,
"inputs": self._get_inputs_list(graph.input_schema),
"inputs": get_inputs_from_schema(graph.input_schema),
"execution_modes": self._get_execution_modes(graph),
},
),
@@ -370,10 +370,6 @@ class RunAgentTool(BaseTool):
session_id=session_id,
)
def _get_inputs_list(self, input_schema: dict[str, Any]) -> list[dict[str, Any]]:
"""Extract inputs list from schema."""
return get_inputs_from_schema(input_schema)
def _get_execution_modes(self, graph: GraphModel) -> list[str]:
"""Get available execution modes for the graph."""
trigger_info = graph.trigger_setup_info
@@ -387,7 +383,7 @@ class RunAgentTool(BaseTool):
suffix: str,
) -> str:
"""Build a message describing available inputs for an agent."""
inputs_list = self._get_inputs_list(graph.input_schema)
inputs_list = get_inputs_from_schema(graph.input_schema)
required_names = [i["name"] for i in inputs_list if i["required"]]
optional_names = [i["name"] for i in inputs_list if not i["required"]]