picoware.system.agent.agent.Agent

class documentation

class Agent:

Constructor: Agent(view_manager, mode, llm, file_path)

Agent that can perform tasks using tools and LLMs.

Method	`__del__`	Cleanup resources on deletion.
Method	`__init__`	Initialize the agent with a mode, LLM, and request file path.
Method	`close`	Close the agent and stop any running tasks.
Method	`run`	Run the agent for a prompt and return the response text.
Method	`run_payload`	Run the agent with a JSON payload and return a structured response.
Method	`run_session`	Run the agent with a session ID and user message, returning a structured response.
Method	`run_session_async`	Start a session on another thread.
Class Variable	`__slots__`	Undocumented
Instance Variable	`http`	Undocumented
Instance Variable	`llm`	Undocumented
Instance Variable	`mode`	Undocumented
Instance Variable	`tools`	Undocumented
Instance Variable	`view_manager`	Undocumented
Property	`file_path`	Get the file path associated with the agent.
Property	`in_progress`	Check if the agent is currently running a task.
Property	`response`	Get the latest response from the agent.
Static Method	`_json_escape`	Escape a string for embedding in JSON.
Static Method	`_mcp_tool_name`	Create a unique model-safe name for an MCP tool.
Static Method	`_sanitize_tool_name`	Replace characters unsupported by common tool-name schemas.
Static Method	`_stream_file_json_escaped`	Stream a file to the destination with JSON escaping.
Method	`_build_request`	Stream the conversation and metadata into the API request file.
Method	`_build_tools`	Build built-in and currently selected MCP tool schemas.
Method	`_conv_append`	Append one message to the conversation file.
Method	`_conv_write_initial`	Write the initial conversation messages to the conversation file.
Method	`_execute_tool`	Execute a built-in or selected MCP tool.
Method	`_parse_tool_arguments`	Parse tool-call arguments defensively into a dict.
Method	`_run_loop`	Run the model/tool loop until a final reply is produced.
Method	`_sanitize_conversation`	Normalize history to user and assistant text messages only.
Method	`_select_mcp_server`	Select an MCP server and expose its discovered tools.
Method	`_should_continue`	Check if the request should continue running.
Method	`_store_response`	Store the latest response for the caller to pick up.
Method	`_write_system_message`	Write the system message to the conversation file.
Instance Variable	`_conv_path`	Undocumented
Instance Variable	`_current_task`	Undocumented
Instance Variable	`_file_path`	Undocumented
Instance Variable	`_lock`	Undocumented
Instance Variable	`_mcp`	Undocumented
Instance Variable	`_mcp_tool_routes`	Undocumented
Instance Variable	`_mcp_tool_schemas`	Undocumented
Instance Variable	`_mem_path`	Undocumented
Instance Variable	`_msg_path`	Undocumented
Instance Variable	`_response`	Undocumented
Instance Variable	`_running`	Undocumented

def __del__(self): ¶

Cleanup resources on deletion.

def __init__(self, view_manager, mode: int = MODE_CHAT, llm: LLM = None, file_path: str = 'picoware/settings/agent_request.json'): ¶

Initialize the agent with a mode, LLM, and request file path.

Parameters
view_manager:`ViewManager`	The view manager for storage and threading.
mode:`int`	The agent mode constant. Defaults to MODE_CHAT.
llm:`LLM`	The LLM client to use. Defaults to None.
file_path:`str`	Path to the API request file. Defaults to "picoware/settings/agent_request.json".

def close(self): ¶

Close the agent and stop any running tasks.

def run(self, topic: str, conversation: list[dict] | None = None, context=None) -> str: ¶

Run the agent for a prompt and return the response text.

Parameters
topic:`str`	The user prompt.
conversation:`list[dict] or None`	Prior message history. Defaults to None.
context:`str or None`	Extra context prepended to the system prompt. Defaults to None.

Returns
`str`	The assistant response text, or an error message.

def run_payload(self, payload: dict) -> dict: ¶

Run the agent with a JSON payload and return a structured response.

Parameters
payload:`dict`	The request payload with message and conversation keys.

Returns
`dict`	The response with status, message, and conversation keys.

def run_session(self, session_id: str, user_message: str) -> dict: ¶

Run the agent with a session ID and user message, returning a structured response.

Parameters
session_id:`str`	The unique identifier for the session.
user_message:`str`	The user's message to process.

Returns
`dict`	The response with status, message, and conversation keys.

def run_session_async(self, session_id: str, user_message: str) -> bool: ¶

Start a session on another thread.

The session result is available from the response property once in_progress returns False.

Parameters
session_id:`str`	The unique identifier for the session.
user_message:`str`	The user's message to process.

Returns
`bool`	True if the session was successfully started, False otherwise.

__slots__: list[str] = ¶

Undocumented

http = ¶

Undocumented

llm = ¶

Undocumented

mode = ¶

Undocumented

tools: list = ¶

Undocumented

view_manager = ¶

Undocumented

@property

file_path: str = ¶

Get the file path associated with the agent.

@property

in_progress: bool = ¶

Check if the agent is currently running a task.

@property

response: str | dict | None = ¶

Get the latest response from the agent.

Returns
`str \| dict \| None`	The session result dict after a session run, the response text otherwise, or None when no run completed.

@staticmethod

def _json_escape(text: str) -> str: ¶

Escape a string for embedding in JSON.

Parameters
text:`str`	The raw string to escape.

Returns
`str`	The JSON-escaped string.

@staticmethod

def _mcp_tool_name(server_id: str, tool_name: str, used_names: list[str]) -> str: ¶

Create a unique model-safe name for an MCP tool.

@staticmethod

def _sanitize_tool_name(name: str) -> str: ¶

Replace characters unsupported by common tool-name schemas.

@staticmethod

def _stream_file_json_escaped(storage, src_path: str, dst_path: str): ¶

Stream a file to the destination with JSON escaping.

Parameters
storage:`Storage`	The storage interface.
src_path:`str`	The source file path.
dst_path:`str`	The destination file path.

def _build_request(self, tools: list[dict]): ¶

Stream the conversation and metadata into the API request file.

Parameters
tools:`list[dict]`	The tool schemas to include in the request.

def _build_tools(self) -> list[dict]: ¶

Build built-in and currently selected MCP tool schemas.

def _conv_append(self, message: dict): ¶

Append one message to the conversation file.

Parameters
message:`dict`	The message to append.

def _conv_write_initial(self, messages: list[dict]): ¶

Write the initial conversation messages to the conversation file.

Parameters
messages:`list[dict]`	The initial messages to store.

def _execute_tool(self, name: str, args: dict): ¶

Execute a built-in or selected MCP tool.

def _parse_tool_arguments(self, raw_args) -> dict: ¶

Parse tool-call arguments defensively into a dict.

Parameters
raw_args:`str or dict`	Raw arguments from the model call.

Returns
`dict`	The parsed arguments, or an empty dict if unparseable.

def _run_loop(self, timeout: int = 300) -> str: ¶

Run the model/tool loop until a final reply is produced.

Returns
`str`	The final assistant text, or an error message.

def _sanitize_conversation(self, conversation: list[dict] | None, max_messages: int = MAX_CONVERSATION_MESSAGES) -> list[dict[str, str]]: ¶

Normalize history to user and assistant text messages only.

Parameters
conversation:`list[dict] or None`	Raw message history. Defaults to None.
max_messages:`int`	Maximum messages to keep. Defaults to MAX_CONVERSATION_MESSAGES.

Returns
`list[dict[str, str]]`	The sanitized message list.

def _select_mcp_server(self, args: dict) -> dict: ¶

Select an MCP server and expose its discovered tools.

def _should_continue(self) -> bool: ¶

Check if the request should continue running.

def _store_response(self, response, done: bool = False): ¶

Store the latest response for the caller to pick up.

Parameters
response:`str or dict`	The response to store.
done:`bool`	Clear the running flag when True. Defaults to False.

def _write_system_message(self, storage): ¶

Write the system message to the conversation file.

Parameters
storage:`Storage`	The storage interface.

_conv_path: str = ¶

Undocumented

_current_task = ¶

Undocumented

_file_path = ¶

Undocumented

_lock = ¶

Undocumented

_mcp = ¶

Undocumented

_mcp_tool_routes: dict = ¶

Undocumented

_mcp_tool_schemas: list = ¶

Undocumented

_mem_path: str = ¶

Undocumented

_msg_path: str = ¶

Undocumented

_response = ¶

Undocumented

_running: bool = ¶

Undocumented