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

.gitignore

@@ -22,3 +22,6 @@ dmypy.json
 
 # misc
 .DS_Store
+
+# Tests
+.*_cache/

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

agntcy_iomapper/__init__.py

@@ -7,6 +7,7 @@
     IOMapperOutput,
     IOModelSettings,
 )
+from .imperative import ImperativeIOMapper
 from .iomapper import (
     AgentIOMapper,
     IOMapperConfig,

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):

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, 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) -> 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

agntcy_iomapper/iomapper.py

@@ -2,21 +2,21 @@
 # SPDX-License-Identifier: Apache-2.0
 import argparse
 import asyncio
-import aiofiles
-import logging
-import jsonschema
 import json
+import logging
 import re
+from typing import ClassVar, TypedDict
 
-from dotenv import load_dotenv, find_dotenv
-from pydantic import Field, model_validator, BaseModel
-from typing import TypedDict, ClassVar
-from typing_extensions import Self
+import aiofiles
+import jsonschema
+from dotenv import find_dotenv, load_dotenv
 from jinja2 import Environment
 from jinja2.sandbox import SandboxedEnvironment
+from pydantic import BaseModel, Field, model_validator
 from pydantic_ai import Agent
+from typing_extensions import Self
 
-from .base import BaseIOMapper, IOModelSettings, IOMapperOutput, IOMapperInput
+from .base import BaseIOMapper, IOMapperInput, IOMapperOutput, IOModelSettings
 from .supported_agents import get_supported_agent
 
 logger = logging.getLogger(__name__)
@@ -76,7 +76,9 @@ def _validate_obj(self) -> Self:
 
 
 class AgentIOMapper(BaseIOMapper):
-    _json_search_pattern: ClassVar[re.Pattern] = re.compile(r"```json\n(.*?)\n```", re.DOTALL)
+    _json_search_pattern: ClassVar[re.Pattern] = re.compile(
+        r"```json\n(.*?)\n```", re.DOTALL
+    )
 
     def __init__(
         self,

agntcy_iomapper/supported_agents.py

@@ -1,6 +1,7 @@
 # SPDX-FileCopyrightText: Copyright (c) 2025 Cisco and/or its affiliates.
 # SPDX-License-Identifier: Apache-2.0
-from typing import Literal, Any
+from typing import Any, Literal
+
 from openai import AsyncAzureOpenAI
 from pydantic_ai import Agent
 from pydantic_ai.models import KnownModelName
@@ -27,16 +28,16 @@ def get_supported_agent(
     Args:
         model_name (SupportedModelName): The name of the model to be used.
             If the name starts with "azure:", an `AsyncAzureOpenAI` client is used.
-        model_args (dict[str, Any], optional): Additional arguments for model 
+        model_args (dict[str, Any], optional): Additional arguments for model
             initialization. Defaults to an empty dictionary.
         **kwargs: Additional keyword arguments passed to the `Agent` constructor.
 
     Returns:
         Agent: An instance of the `Agent` class configured with the specified model.
 
     Notes:
-        - The `pydantic-ai` package does not currently pass `model_args` to the 
-          inferred model in the constructor, but this behavior might change in 
+        - The `pydantic-ai` package does not currently pass `model_args` to the
+          inferred model in the constructor, but this behavior might change in
           the future.
     """
     if model_name.startswith("azure:"):

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)
+
 ## 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,

docs/usage.md

@@ -0,0 +1,63 @@
+# 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)

examples/Makefile

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