aidputils.agents.toolkit.tool_helper

aidputils.agents.toolkit.tool_helper.create_tool_function(tool_name: str)

Create a LangGraph-compatible wrapper that invokes a registered toolkit tool.

The returned callable injects the selected tool_name into the runtime parameters and delegates execution to aidputils.agents.tools.utils.call_tool_by_name. This allows the same registry-backed tool configuration to be reused by both the toolkit runtime and generated StructuredTool objects.

aidputils.agents.toolkit.tool_helper.derive_primitive_type(type_str)

Map a tool-spec type string to a Python runtime type.

This helper normalizes loose type spellings coming from tool metadata so dynamic Pydantic schemas can be constructed consistently.

Supports: - Primitives: str/string, int/integer, float/double/number/numeric, bool/boolean, bytes, bytearray - Nulls: none, null - Collections: list/array/sequence/iterable, dict/dictionary/map/mapping, set, frozenset, tuple - Basic generic-like patterns: “list[int]”, “set[str]”, “tuple[int, str]” or “list of int”

In these cases, we return the base container type (list/set/tuple/dict).

aidputils.agents.toolkit.tool_helper.create_langgraph_tool(tool)

Convert a toolkit tool configuration into a StructuredTool instance.

The resulting tool uses the global tool_registry so downstream execution can look up the original tool metadata and invoke the correct implementation at runtime.

aidputils.agents.toolkit.tool_helper.create_langgraph_custom_tool(tool)

Create a StructuredTool backed by a user-provided Python callable.

This path is intended for custom tools that do not rely on the standard toolkit registry invocation flow but should still appear as first-class LangChain tools.

aidputils.agents.toolkit.tool_helper.create_tool_schema(tool)

Build a dynamic Pydantic input model from a toolkit tool parameter definition.

aidputils.agents.toolkit.tool_helper.build_args_schema_from_json_schema(schema: dict | None, model_name: str = 'MCPToolInput')

Convert a JSON Schema (as provided by mcp.types.Tool.inputSchema) into a Pydantic model class. - Supports basic ‘properties’ and ‘required’ handling - Maps primitive types and falls back to Any for unknowns

aidputils.agents.toolkit.tool_helper.mcp_tool_to_structured_tool(mcp_tool: mcp.types.Tool, coroutine: Callable[[...], Awaitable[Any]] | None = None) → langchain_core.tools.StructuredTool

Convert an mcp.types.Tool into a LangChain StructuredTool. - Name/description taken from the MCP tool - Args schema built from MCP tool’s inputSchema JSON schema - ‘func’ is the callable to execute when the tool is invoked; if not provided,

a default invoker returns a JSON-serializable envelope with the tool name and arguments.

Example usage:

st = mcp_tool_to_structured_tool(mcp_tool, func=lambda **kw: client.call_tool(…))

aidputils.agents.toolkit.tool_helper.build_mcp_invoker_factory(client, timeout_secs: float | None = None, runtime_params: dict = None)

Build an invoker factory for remotely hosted MCP tools.

Each generated coroutine merges configured default arguments, resolves auth context, and routes the request through the provided MCP client.

Defaults merging (scoped by server_name to avoid cross-server conflicts): - allowed_tools entries may include:

• name/toolName: specific tool name

• argOverrides/arg_overrides: dict of default argument values

• server_name/serverName (optional): to explicitly scope the entry

• Merge order per invocation: per-tool defaults (for this server) -> user kwargs (user wins)

aidputils.agents.toolkit.tool_helper.build_args_schema_from_json_schema_with_overrides(schema: dict | None, overrides: dict | None, model_name: str = 'MCPToolInput')

Build a Pydantic model from a JSON Schema, applying argOverrides:

• For every attribute present in overrides:

  • remove it from input schema ‘properties’

  • remove it from ‘required’ (so it is not requested from the LLM)

• Do NOT treat overrides as defaults in the schema.

• Then build a Pydantic model from the pruned schema.

aidputils.agents.toolkit.tool_helper.build_structured_tools_from_allowed_mcp_tools(allowed_tools: list[dict], server_name: str | None = None, endpoint: str | None = None, transport: str = 'streamable_http', auth: dict | None = None, headers: dict | None = None, invoker_factory: Callable[[str], Callable[[...], Awaitable[Any]]] | None = None) → list[langchain_core.tools.StructuredTool]

Given an array of AllowedToolDetails (from DP spec), return a list of StructuredTool.

AllowedToolDetails schema (relevant parts):

• instruction: optional custom instruction to override description

• argOverrides: map[string]string default values for tool params

• tool: McpToolObject { name, description, inputSchema }

Behavior:

• Build args_schema from tool.inputSchema and apply argOverrides:

  • required fields present in argOverrides are made optional with provided defaults

  • non-required fields prefer override as default

• Description is overridden by ‘instruction’ when provided, else tool.description

• If invoker_factory not provided and server_name/endpoint are provided, this function will:

  • Construct an MCP client internally (not stored by caller) and

  • Build an invoker_factory (via build_mcp_invoker_factory) so tool coroutines call client.call_tool.

  No client storage/cleanup is required by the caller.

• If invoker_factory is provided explicitly, it will be used directly.

Returns:

list[StructuredTool]