feat: add ability to imperatively defining the mapping between input ouput

Author: CostaRegi

Diff for: README.md

@@ -15,15 +15,15 @@ Communication between agents is not possible if there are discrepancies between
 
 Ensuring that agents are semantically compatible, i.e., the output of the one agent contains the information needed
 by later agents, is an problem of composition or planning in the application. This project, the IO Mapper Agent,
-addresses level 2 and 3 compatibility.  It is a component, implemented as an agent, that can make use of an LLM
+addresses level 2 and 3 compatibility. It is a component, implemented as an agent, that can make use of an LLM
 to transform the output of one agent to become compatible to the input of another agent. Note that this may mean
 many different things, for example:
 
-* JSON structure transcoding: A JSON dictionary needs to be remapped into another JSON dictionary
-* Text summarisation: A  text needs to be summarised or some information needs to be removed
-* Text translation: A text needs to be translated from one language to another
-* Text manipulation: Part of the information of one text needs to be reformulated into another text
-* Any combination of the above
+- JSON structure transcoding: A JSON dictionary needs to be remapped into another JSON dictionary
+- Text summarisation: A text needs to be summarised or some information needs to be removed
+- Text translation: A text needs to be translated from one language to another
+- Text manipulation: Part of the information of one text needs to be reformulated into another text
+- Any combination of the above
 
 The IO mapper Agent can be fed the schema definitions of inputs and outputs as defined by the [Agent Connect Protocol](https://github.com/agntcy/acp-spec).
 
@@ -43,6 +43,9 @@ To get a local copy up and running follow these simple steps.
 
 ## Usage
 
+Learn how to use our different Mappers
+[USAGE.md](_usage.md)
+
 ## Contributing
 
 Contributions are what make the open source community such an amazing place to

Diff for: agntcy_iomapper/__init__.py

@@ -12,3 +12,7 @@
     IOMapperConfig,
     IOModelArgs,
 )
+
+from .imperative import ImperativeIOMapper
+
+___all__ = [ "ImperativeIOMapper", "AgentIOMapper", "IOMapperOutput", "IOMapperInput"]

Diff for: agntcy_iomapper/base.py

@@ -1,11 +1,11 @@
 # SPDX-FileCopyrightText: Copyright (c) 2025 Cisco and/or its affiliates.
 # SPDX-License-Identifier: Apache-2.0
-from abc import abstractmethod, ABC
-from pydantic import BaseModel, model_validator, Field
-from typing import Any
+from abc import ABC, abstractmethod
+from typing import Any, TypedDict
+
 from openapi_pydantic import Schema
+from pydantic import BaseModel, Field, model_validator
 from typing_extensions import Self
-from typing import TypedDict
 
 
 class ArgumentsDescription(BaseModel):

Diff for: agntcy_iomapper/imperative.py

@@ -0,0 +1,136 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025 Cisco and/or its affiliates.
+# SPDX-License-Identifier: Apache-2.0"
+"""
+The deterministic I/O mapper is a component
+designed to translate specific inputs into
+corresponding outputs in a predictable and consistent manner.
+When configured with a JSONPath definition,
+this mapper utilizes the JSONPath query language
+to extract data from JSON-formatted input,
+transforming it into a structured output based on predefined rules.
+The deterministic nature of the mapper ensures that given the same input and
+JSONPath configuration, the output will always be the same,
+providing reliability and repeatability.
+This is particularly useful in scenarios where
+consistent data transformation is required.
+"""
+
+import json
+import logging
+from typing import Any, Callable, Dict, Union
+
+import jsonschema
+from jsonpath_ng.ext import parse
+
+from agntcy_iomapper.base import BaseIOMapper, IOMapperInput, IOMapperOutput
+
+logger = logging.getLogger(__name__)
+
+
+class ImperativeIOMapper(BaseIOMapper):
+    field_mapping: Dict[str, Union[str, Callable]]
+    """A dictionary for where the keys are fields of the output object
+    and values are JSONPath (strings) representing how the mapping
+    """
+
+    def __init__(self, field_mapping: Dict[str, Union[str, Callable]]) -> None:
+        super().__init__()
+        self.field_mapping = field_mapping
+
+    def invoke(self, input: IOMapperInput) -> IOMapperOutput | None:
+        if input.data is None:
+            return None
+        if self.field_mapping is None:
+            return IOMapperOutput(data=input.data)
+
+        data = self._imperative_map(input)
+        return IOMapperOutput(data=data)
+
+    def ainvoke(self, input: IOMapperInput) -> IOMapperOutput | None:
+        return self.invoke(input)
+
+    def _imperative_map(self, input_definition: IOMapperInput) -> Any:
+        """
+        Converts input data to a desired output type.
+
+        This function attempts to convert the provided data into the specified
+        target type. It performs validation using a JSON schema and raises a
+        ValidationError if the data does not conform to the expected schema for
+        the target type.
+
+        Parameters:
+        ----------
+        data : Any
+            The input data to be converted. This can be of any type.
+        Returns:
+        -------
+        Any
+            The converted data in the desired output type.
+        Raises:
+        ------
+        ValidationError
+            If the input data does not conform to the expected schema for the
+            target type.
+        Notes:
+        -----
+        The function assumes that the caller provides a valid `input_schema`.
+        Unsupported target types should be handled as needed within the function.
+        """
+        data = input_definition.data
+        input_schema = input_definition.input.json_schema
+
+        jsonschema.validate(
+            instance=data,
+            schema=input_schema.model_dump(exclude_none=True, mode="json"),
+        )
+
+        mapped_output = {}
+
+        for output_field, json_path_or_func in self.field_mapping.items():
+            if isinstance(json_path_or_func, str):
+                jsonpath_expr = parse(json_path_or_func)
+                match = jsonpath_expr.find(data)
+                expect_value = match[0].value if match else None
+            elif callable(json_path_or_func):
+                expect_value = json_path_or_func(data)
+            else:
+                raise TypeError(
+                    "Mapping values must be strings (JSONPath) or callables (functions)."
+                )
+
+            self._set_jsonpath(mapped_output, output_field, expect_value)
+        jsonschema.validate(
+            instance=mapped_output,
+            schema=input_definition.output.json_schema.model_dump(
+                exclude_none=True, mode="json"
+            ),
+        )
+        # return a serialized version of the object
+        return json.dumps(mapped_output)
+
+    def _set_jsonpath(
+        self, data: dict[str, Any], path: str, value: Any
+    ) -> dict[str, Any]:
+        """set value for field based on its json path
+        Args:
+            data: Data so far
+            path: the json path
+            value: the value to set the json path to
+        Returns:
+        -----
+        dict[str,Any]
+            The mapped filed with the value
+        """
+        copy_data: dict[str, Any] = data
+        # Split the path into parts and remove the leading root
+        parts = path.strip("$.").split(".")
+        # Add value to corresponding path
+        for part in parts[:-1]:
+            if part not in copy_data:
+                copy_data[part] = {}
+
+            copy_data = copy_data[part]
+
+        copy_data[parts[-1]] = value
+
+        return copy_data

Diff for: docs/README.md

@@ -12,15 +12,15 @@ Communication between agents is not possible if there are discrepancies between
 
 Ensuring that agents are semantically compatible, i.e., the output of the one agent contains the information needed
 by later agents, is an problem of composition or planning in the application. This project, the IO Mapper Agent,
-addresses level 2 and 3 compatibility.  It is a component, implemented as an agent, that can make use of an LLM
+addresses level 2 and 3 compatibility. It is a component, implemented as an agent, that can make use of an LLM
 to transform the output of one agent to become compatible to the input of another agent. Note that this may mean
 many different things, for example:
 
-* JSON structure transcoding: A JSON dictionary needs to be remapped into another JSON dictionary
-* Text summarisation: A  text needs to be summarised or some information needs to be removed
-* Text translation: A text needs to be translated from one language to another
-* Text manipulation: Part of the information of one text needs to be reformulated into another text
-* Any combination of the above
+- JSON structure transcoding: A JSON dictionary needs to be remapped into another JSON dictionary
+- Text summarisation: A text needs to be summarised or some information needs to be removed
+- Text translation: A text needs to be translated from one language to another
+- Text manipulation: Part of the information of one text needs to be reformulated into another text
+- Any combination of the above
 
 The IO mapper Agent can be fed the schema definitions of inputs and outputs as defined by the [Agent Connect Protocol](https://github.com/agntcy/acp-spec).
 
@@ -40,6 +40,9 @@ To get a local copy up and running follow these simple steps.
 
 ## Usage
 
+Learn how to use our different Mappers
+[USAGE.md](_usage.md)
+
 ## Contributing
 
 Contributions are what make the open source community such an amazing place to
@@ -61,7 +64,7 @@ Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
 You may obtain a copy of the License at
 
-   http://www.apache.org/licenses/LICENSE-2.0
+http://www.apache.org/licenses/LICENSE-2.0
 
 Unless required by applicable law or agreed to in writing, software
 distributed under the License is distributed on an "AS IS" BASIS,

Diff for: docs/_usage.md

@@ -0,0 +1,62 @@
+# Usage
+
+### Use Agent IO Mapper
+
+:TODO
+
+### Use Determenistic
+
+The code snippet below illustrates a fully functional deterministic mapping that transforms the output of one agent into input for a second agent. The code for the agents is omitted.
+
+```python
+ #define schema for the origin agent
+ input_schema = {"question": {"type": "string"}}
+
+ #define schema to witch the input should be converted to
+ output_schema = {
+     "quiz": {
+         "type": "object",
+         "properties": {
+             "prof_question": {"type": "string"},
+             "due_date": {"type": "string"},
+         },
+     }
+ }
+
+ #the mapping object using jsonpath, note: the value of the mapping can be either a jsonpath or a function
+ mapping_object = {
+     "prof_question": "$.question",
+     "due_date": lambda _: datetime.now().strftime("%x"),
+ }
+
+ input = IOMapperInput(
+     input=ArgumentsDescription(
+         json_schema=Schema.model_validate(input_schema)
+     ),
+     output=ArgumentsDescription(
+         json_schema=Schema.model_validate(output_schema)
+     ),
+     data={"question": output_prof},
+ )
+ #instantiate the mapper
+ imerative_mapp = ImperativeIOMapper(
+     field_mapping=mapping_object,
+ )
+ #get the mapping result and send to the other agent
+ mapping_result = imerative_mapp.invoke(input=input)
+
+
+
+```
+
+### Use Examples
+
+1. To run the examples we strongly recommend that a [virtual environment is created](https://packaging.python.org/en/latest/guides/installing-using-pip-and-virtual-environments/)
+2. Install the requirements file
+3. from within examples folder run:
+
+```sh make run_imperative_example
+
+[GitHub](https://github.com/agntcy/iomapper-agnt/)
+[Get Started](#getting-started)
+```

Diff for: examples/Makefile

@@ -0,0 +1,5 @@
+
+EXAMPLES ?= .
+
+run_imperative_example:
+	python $(EXAMPLES)/imperative.py