aidputils.agents.toolkit.agent_helper

Helper methods for initializing Oracle Cloud GenAI LLMs.

aidputils.agents.toolkit.agent_helper.init_oci_llm(llm_conf: OCIAIConf)

Initialize GuardedChatOCIGenAI language model instance using the provided OCIAIConf.

Parameters:

llm_conf (OCIAIConf) – The configuration object specifying model, provider, endpoint, compartment, authentication profile, model args, and optional guardrails config.

Returns:

Instantiated language model object ready for use with the specified settings.

Return type:

GuardedChatOCIGenAI

aidputils.agents.toolkit.agent_helper.get_client(llm_conf)

aidputils.agents.toolkit.agent_helper.pre_tool_setup(**kwargs)

aidputils.agents.toolkit.agent_helper.pre_invoke_setup(**kwargs) → dict

aidputils.agents.toolkit.agent_helper.post_tool_setup(token, mcp_clients: List[MCPHTTPClient] | None = None, kwargs: dict | None = None)

Synchronous cleanup after tool execution.

• Always resets the auth context using the provided token.

• If mcp_clients and session_id are provided, synchronously stops each MCP client session, bridging async calls via _run_in_event_loop to work whether an event loop is running or not.

async aidputils.agents.toolkit.agent_helper.parse_stream_response(stream)

Parse and yield message chunks from a LangGraph streaming response. Yields only discrete message chunks (BaseMessage or equivalent) from events.

async aidputils.agents.toolkit.agent_helper.parse_stream_response_for_supervisor(stream: AsyncIterator[object], *, supervisor_name: str, include_intermediate: bool = False) → AsyncIterator[object]

UI-friendly stream parser.

The default LangGraph stream_mode=”messages” will emit all messages from all nodes, including worker outputs and handoff markers. Many chat UIs then concatenate those into one transcript, causing duplicated/verbose output and slow rendering.

This generator filters the stream to yield only end-user facing messages.

Rules: - Drop ToolMessage (tool chatter) - Drop any handoff marker messages - By default, only emit AI messages from the supervisor node (by .name) - Deduplicate repeated messages

Parameters:

• stream – The raw async stream from agent.astream(…)

• supervisor_name – The graph node name used for the supervisor agent.

• include_intermediate – If True, also allow non-supervisor AI messages (still filtered for handoff/tool chatter). Default False.

async aidputils.agents.toolkit.agent_helper.stream_messages(self, input, config: dict | None = None, kwargs: dict | None = None) → AsyncGenerator[Any, None]

Stream messages as they are generated by the agent. Yields BaseMessage objects as new messages are added by nodes. Ensures the auth context (ContextVar) remains active during streaming and is cleaned up after.

class aidputils.agents.toolkit.agent_helper.FlowInvokeSetup(*, token, config: dict, kwargs: dict)

Bases: object

Backward-compatible wrapper for pre/post helper methods.

This class exists to provide a single, higher-level API that composes:

• pre_invoke_setup(**kwargs) -> config

• pre_tool_setup(**kwargs) -> token

• post_tool_setup(token, mcp_clients=…, kwargs=…) + auth context reset

Existing functions remain unchanged for backward compatibility.

cleanup(mcp_clients: List[MCPHTTPClient] | None = None)

Run the same cleanup as post_tool_setup and reset auth context.

Idempotent: safe to call multiple times.

aidputils.agents.toolkit.agent_helper.setup_tool_invoke(**kwargs) → FlowInvokeSetup

Merged helper that replaces calling pre_* and post_* manually.

Example

setup = setup_tool_invoke(**kwargs) try:

result = await agent.ainvoke(input=message, config=setup.config) return result

finally:

setup.cleanup(mcp_clients=mcp_clients)

Returns:

contains the invocation config and a cleanup() method.

Return type:

FlowInvokeSetup

aidputils.agents.toolkit.agent_helper.flow_setup(*, mcp_clients: List[MCPHTTPClient] | None = None, **kwargs)

Context-manager wrapper around setup_tool_invoke().

This enables simplified user code:

with flow_setup(mcp_clients=mcp_clients, **kwargs) as config:

agent_response = await agent.ainvoke(input=message, config=config)

Yields:

dict – The invocation config (same as pre_invoke_setup).

aidputils.agents.toolkit.agent_helper.fill_template(template: str, variables: Dict | None) → str

aidputils.agents.toolkit.agent_helper.fill_template_from_session_vars(template: str, variables: Dict | None = None) → str

Build a prompt by merging variables from session context with provided variables and defaults. - template uses Python str.format placeholders, supports dotted keys like {sessionvariables.tone} - variables is optional additional override dict; its keys take precedence over session vars

aidputils.agents.toolkit.agent_helper.create_dynamic_sys_prompt_middleware(template: str)

aidputils.agents.toolkit.agent_helper.create_get_dynamic_messages(system_prompt_template: str)

Factory that returns a get_dynamic_messages(state) function compatible with LangGraph’s prompt callback. The returned function:

• Renders a SystemMessage from system_prompt_template using session vars merged with provided defaults

• Prepends that SystemMessage to the existing history in state[‘messages’]

• Returns [SystemMessage] + history