ReMe/reme2/steps/runtime_response.py
huangsen 1852d05f7b feat(file_graph): add optional paths parameter to get_nodes method
Add support for `paths=None` in `get_nodes` method to return all real
nodes as the graph's "scan everything" entry point. When `paths` is
explicitly passed as an empty list, returns empty list. Virtual
placeholders are filtered out in both cases.

BREAKING CHANGE: The `get_nodes` method signature changed from
requiring a list of paths to accepting an optional list or None.

refactor(memory): move runtime_response imports to steps package

Move runtime_response imports from memory package to steps package to
improve code organization and reduce circular dependencies.

refactor(config): update obsidian configuration table structure

Update the obsidian configuration markdown table to improve clarity
and fix incorrect section headings. Change 'stat/list' to 'file' and
'return specific tag info' to 'tags' section with proper commands.

chore(memory): remove deprecated toolkit modules

Remove deprecated agent_toolkit.py, file_toolkit.py, and
graph_toolkit.py modules as functionality has been moved to steps
package.

feat(steps): add new CRUD file operations

Add new steps for file operations including file_download and
file_list operations with proper path resolution and filtering
capabilities.
2026-05-15 14:55:08 +08:00

51 lines
1.6 KiB
Python

"""Helpers for serializing Step output onto `RuntimeContext.response`
and into `agentscope.tool.ToolResponse`.
Lives next to `runtime_context.py` because both are about the BaseStep
interface — the response side, specifically. Used by every Step that
returns a JSON-shaped payload (agent toolkit, lint toolkit, the three
memory services).
"""
from __future__ import annotations
import json
from datetime import date, datetime
from typing import Any
from agentscope.message import TextBlock
from agentscope.tool import ToolResponse
def _to_jsonable(value):
if isinstance(value, (datetime, date)):
return value.isoformat()
if isinstance(value, dict):
return {k: _to_jsonable(v) for k, v in value.items()}
if isinstance(value, (list, tuple, set)):
return [_to_jsonable(v) for v in value]
return value
def _set_answer(context, payload) -> None:
context.response.answer = json.dumps(_to_jsonable(payload), ensure_ascii=False, indent=2)
def _tool_response(
op: str,
ok: bool,
payload: Any,
audit: list[dict] | None = None,
) -> ToolResponse:
"""Wrap a tool-method result as `ToolResponse` and optionally
append an audit row. Shared by every BaseStep's tool-method
surface so each toolkit doesn't reinvent serialization."""
if audit is not None:
entry = {"op": op, "ok": ok}
if isinstance(payload, dict):
entry.update(payload)
else:
entry["result"] = payload
audit.append(entry)
text = json.dumps(_to_jsonable(payload), ensure_ascii=False, indent=2)
return ToolResponse(content=[TextBlock(type="text", text=text)])