robocorp[patch]: Fix nested arguments descriptors and tool names

libs/partners/robocorp/README.md

@@ -1,6 +1,7 @@
 # langchain-robocorp
 
-This package contains the LangChain integrations for [Robocorp](https://github.com/robocorp/robocorp).
+This package contains the LangChain integrations for [Robocorp Action Server](https://github.com/robocorp/robocorp).
+Action Server enables an agent to execute actions in the real world. 
 
 ## Installation
 

libs/partners/robocorp/langchain_robocorp/_common.py

@@ -1,6 +1,7 @@
 from dataclasses import dataclass
-from typing import List, Tuple
+from typing import Any, Dict, List, Tuple, Union
 
+from langchain_core.pydantic_v1 import BaseModel, Field, create_model
 from langchain_core.utils.json_schema import dereference_refs
 
 
@@ -72,28 +73,6 @@ def reduce_endpoint_docs(docs: dict) -> dict:
     )
 
 
-def get_required_param_descriptions(endpoint_spec: dict) -> str:
-    """Get an OpenAPI endpoint required parameter descriptions"""
-    descriptions = []
-
-    schema = (
-        endpoint_spec.get("requestBody", {})
-        .get("content", {})
-        .get("application/json", {})
-        .get("schema", {})
-    )
-    properties = schema.get("properties", {})
-
-    required_fields = schema.get("required", [])
-
-    for key, value in properties.items():
-        if "description" in value:
-            if value.get("required") or key in required_fields:
-                descriptions.append(value.get("description"))
-
-    return ", ".join(descriptions)
-
-
 type_mapping = {
     "string": str,
     "integer": int,
@@ -105,25 +84,66 @@ def get_required_param_descriptions(endpoint_spec: dict) -> str:
 }
 
 
-def get_param_fields(endpoint_spec: dict) -> dict:
-    """Get an OpenAPI endpoint parameter details"""
-    fields = {}
-
-    schema = (
+def get_schema(endpoint_spec: dict) -> dict:
+    return (
         endpoint_spec.get("requestBody", {})
         .get("content", {})
         .get("application/json", {})
         .get("schema", {})
     )
+
+
+def create_field(schema: dict, required: bool) -> Tuple[Any, Any]:
+    """
+    Creates a Pydantic field based on the schema definition.
+    """
+    field_type = type_mapping.get(schema.get("type", "string"), str)
+    description = schema.get("description", "")
+
+    # Handle nested objects
+    if schema["type"] == "object":
+        nested_fields = {
+            k: create_field(v, k in schema.get("required", []))
+            for k, v in schema.get("properties", {}).items()
+        }
+        model_name = schema.get("title", "NestedModel")
+        nested_model = create_model(model_name, **nested_fields)  # type: ignore
+        return nested_model, Field(... if required else None, description=description)
+
+    # Handle arrays
+    elif schema["type"] == "array":
+        item_type, _ = create_field(schema["items"], required=True)
+        return List[item_type], Field(  # type: ignore
+            ... if required else None, description=description
+        )
+
+    # Other types
+    return field_type, Field(... if required else None, description=description)
+
+
+def get_param_fields(endpoint_spec: dict) -> dict:
+    """Get an OpenAPI endpoint parameter details"""
+    schema = get_schema(endpoint_spec)
     properties = schema.get("properties", {})
     required_fields = schema.get("required", [])
 
+    fields = {}
     for key, value in properties.items():
-        details = {
-            "description": value.get("description", ""),
-            "required": key in required_fields,
-        }
-        field_type = type_mapping[value.get("type", "string")]
-        fields[key] = (field_type, details)
+        is_required = key in required_fields
+        field_info = create_field(value, is_required)
+        fields[key] = field_info
 
     return fields
+
+
+def model_to_dict(
+    item: Union[BaseModel, List, Dict[str, Any]],
+) -> Any:
+    if isinstance(item, BaseModel):
+        return item.dict()
+    elif isinstance(item, dict):
+        return {key: model_to_dict(value) for key, value in item.items()}
+    elif isinstance(item, list):
+        return [model_to_dict(element) for element in item]
+    else:
+        return item

libs/partners/robocorp/langchain_robocorp/_prompts.py

@@ -1,11 +1,10 @@
-# flake8: noqa
-TOOLKIT_TOOL_DESCRIPTION = """{description}. The tool must be invoked with a complete sentence starting with "{name}" and additional information on {required_params}."""
-
-
-API_CONTROLLER_PROMPT = """You are turning user input into a json query for an API request tool.
+API_CONTROLLER_PROMPT = (
+    "You are turning user input into a json query"
+    """ for an API request tool.
 
 The final output to the tool should be a json string with a single key "data".
-The value of "data" should be a dictionary of key-value pairs you want to POST to the url.
+The value of "data" should be a dictionary of key-value pairs you want """
+    """to POST to the url.
 Always use double quotes for strings in the json string.
 Always respond only with the json object and nothing else.
 
@@ -16,3 +15,4 @@
 
 User Input: {input}
 """
+)

libs/partners/robocorp/langchain_robocorp/toolkits.py

@@ -20,12 +20,11 @@
 
 from langchain_robocorp._common import (
     get_param_fields,
-    get_required_param_descriptions,
+    model_to_dict,
     reduce_openapi_spec,
 )
 from langchain_robocorp._prompts import (
     API_CONTROLLER_PROMPT,
-    TOOLKIT_TOOL_DESCRIPTION,
 )
 
 MAX_RESPONSE_LENGTH = 5000
@@ -156,17 +155,9 @@ def get_tools(
             if not endpoint.startswith("/api/actions"):
                 continue
 
-            summary = docs["summary"]
-
-            tool_description = TOOLKIT_TOOL_DESCRIPTION.format(
-                name=summary,
-                description=docs.get("description", summary),
-                required_params=get_required_param_descriptions(docs),
-            )
-
             tool_args: ToolArgs = {
-                "name": f"robocorp_action_server_{docs['operationId']}",
-                "description": tool_description,
+                "name": docs["operationId"],
+                "description": docs["description"],
                 "callback_manager": callback_manager,
             }
 
@@ -218,16 +209,17 @@ def _get_structured_tool(
         self, endpoint: str, docs: dict, tools_args: ToolArgs
     ) -> BaseTool:
         fields = get_param_fields(docs)
+        _DynamicToolInputSchema = create_model("DynamicToolInputSchema", **fields)
 
-        def create_function(endpoint: str) -> Callable:
-            def func(**data: dict[str, Any]) -> str:
-                return self._action_request(endpoint, **data)
+        def dynamic_func(**data: dict[str, Any]) -> str:
+            return self._action_request(endpoint, **model_to_dict(data))
 
-            return func
+        dynamic_func.__name__ = tools_args["name"]
+        dynamic_func.__doc__ = tools_args["description"]
 
         return StructuredTool(
-            func=create_function(endpoint),
-            args_schema=create_model("DynamicToolInputSchema", **fields),
+            func=dynamic_func,
+            args_schema=_DynamicToolInputSchema,
             **tools_args,
         )
 

libs/partners/robocorp/pyproject.toml

@@ -1,7 +1,7 @@
 [tool.poetry]
 name = "langchain-robocorp"
-version = "0.0.4"
-description = "An integration package connecting Robocorp and LangChain"
+version = "0.0.5"
+description = "An integration package connecting Robocorp Action Server and LangChain"
 authors = []
 readme = "README.md"
 repository = "https://github.com/langchain-ai/langchain"