Skip to content

IO Mapper Agent

Contributor Covenant

About the IO Mapper Agent

When connecting agents in an application, the output of an agent needs to be compatible with the input of the agent that is connected to it. This compatibility needs to be guaranteed at three different levels:

  1. Transport level: the two agents need to use the same transport protocol.
  2. Format level: the two agents need to carry information using the same format (for example, the same JSON data structures).
  3. Semantic level: the two agents need to “talk about the same thing”.

Communication between agents is not possible if there are discrepancies between the agents at any of these levels.

Ensuring that agents are semantically compatible, that is, the output of the one agent contains the information needed by later agents, is an problem of composition or planning in the application. 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 to transform the output of one agent to become compatible to the input of another agent. This can mean many different things:

  • 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.
  • A 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>_.

Getting Started

Prerequisites

Use in your project

To install the IO Mapper Agent, run the following command:

pip install agntcy-iomapper

To get a local copy up and running, follow the steps below.

Clone the repository

git clone https://github.com/agntcy/iomapper-agnt.git

Install dependecies

poetry install

Usage

There are several different ways to leverage the IO Mapper functions in Python. There is an How to use the Agent IO mapping using models that can be invoked on different AI platforms and an imperative interface that does deterministic JSON remapping without using any AI models.

Key Features

The IO Mapper Agent uses an LLM to transform the inputs (typically the output of an agent) to match the desired output (typically the input of another agent). As such, it additionally supports specifying the model prompts for the translation. The configuration object provides a specification for the system and default user prompts:

This project supports specifying model interactions using LangGraph.

How to use the Agent IO mapping

Note

For each example, the detailed process of creating agents and configuring the respective multi-agent software is omitted. Instead, only the essential steps for configuring and integrating the IO mapper agent are presented.

LangGraph

We support usages with both LangGraph state defined with TypedDict or as a Pydantic object

Entities

IOMappingAgentMetadata

Bases: BaseModel

Show JSON schema: 
{
  "$defs": {
    "FieldMetadata": {
      "properties": {
        "json_path": {
          "description": "A json path to the field in the object",
          "title": "Json Path",
          "type": "string"
        },
        "description": {
          "description": "A description of what the field represents",
          "title": "Description",
          "type": "string"
        },
        "examples": {
          "anyOf": [
            {
              "items": {
                "type": "string"
              },
              "type": "array"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "A list of examples that represents how the field in json_path is normaly populated",
          "title": "Examples"
        }
      },
      "required": [
        "json_path",
        "description"
      ],
      "title": "FieldMetadata",
      "type": "object"
    }
  },
  "properties": {
    "input_fields": {
      "description": "an array of json paths representing fields to be used by io mapper in the mapping",
      "items": {
        "anyOf": [
          {
            "type": "string"
          },
          {
            "$ref": "#/$defs/FieldMetadata"
          }
        ]
      },
      "title": "Input Fields",
      "type": "array"
    },
    "output_fields": {
      "description": "an array of json paths representing firlds to be used by io mapper in the result",
      "items": {
        "anyOf": [
          {
            "type": "string"
          },
          {
            "$ref": "#/$defs/FieldMetadata"
          }
        ]
      },
      "title": "Output Fields",
      "type": "array"
    },
    "input_schema": {
      "anyOf": [
        {
          "additionalProperties": true,
          "type": "object"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "defines the schema for the input data",
      "title": "Input Schema"
    },
    "output_schema": {
      "anyOf": [
        {
          "additionalProperties": true,
          "type": "object"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "defines the schema for result of the mapping",
      "title": "Output Schema"
    },
    "output_description_prompt": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "A prompt structured using a Jinja template that will be used by the llm for a better mapping",
      "title": "Output Description Prompt"
    },
    "field_mapping": {
      "anyOf": [
        {
          "additionalProperties": {
            "type": "string"
          },
          "type": "object"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "A dictionary representing how the imperative mapping should be done where the keys are fields of the output object and values are JSONPath (strings)",
      "title": "Field Mapping"
    }
  },
  "required": [
    "input_fields",
    "output_fields"
  ],
  "title": "IOMappingAgentMetadata",
  "type": "object"
}

Fields:

Source code in .venv/lib/python3.13/site-packages/agntcy_iomapper/base/models.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
class IOMappingAgentMetadata(BaseModel):
    input_fields: List[Union[str, FieldMetadata]] = Field(
        ...,
        description="an array of json paths representing fields to be used by io mapper in the mapping",
    )
    output_fields: List[Union[str, FieldMetadata]] = Field(
        ...,
        description="an array of json paths representing firlds to be used by io mapper in the result",
    )
    input_schema: Optional[dict[str, Any]] = Field(
        default=None, description="defines the schema for the input data"
    )
    output_schema: Optional[dict[str, Any]] = Field(
        default=None, description="defines the schema for result of the mapping"
    )
    output_description_prompt: Optional[str] = Field(
        default=None,
        description="A prompt structured using a Jinja template that will be used by the llm for a better mapping",
    )
    field_mapping: Optional[dict[str, Union[str, Callable]]] = Field(
        default=None,
        description="A dictionary representing how the imperative mapping should be done where the keys are fields of the output object and values are JSONPath (strings)",
    )

field_mapping = None pydantic-field

A dictionary representing how the imperative mapping should be done where the keys are fields of the output object and values are JSONPath (strings)

input_fields pydantic-field

an array of json paths representing fields to be used by io mapper in the mapping

input_schema = None pydantic-field

defines the schema for the input data

output_description_prompt = None pydantic-field

A prompt structured using a Jinja template that will be used by the llm for a better mapping

output_fields pydantic-field

an array of json paths representing firlds to be used by io mapper in the result

output_schema = None pydantic-field

defines the schema for result of the mapping

IOMappingAgent

Bases: BaseModel

This class exposes all The IOMappingAgent class is designed for developers building sophisticated multi-agent software that require seamless integration and interaction between the different agents and workflow steps. By utilizing the methods provided, developers can construct complex workflows and softwares. The IOMappingAgent class is intended to serve as a foundational component in applications requiring advanced IO mapping agents in multi-agent systems.

Show JSON schema: 
{
  "$defs": {
    "FieldMetadata": {
      "properties": {
        "json_path": {
          "description": "A json path to the field in the object",
          "title": "Json Path",
          "type": "string"
        },
        "description": {
          "description": "A description of what the field represents",
          "title": "Description",
          "type": "string"
        },
        "examples": {
          "anyOf": [
            {
              "items": {
                "type": "string"
              },
              "type": "array"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "A list of examples that represents how the field in json_path is normaly populated",
          "title": "Examples"
        }
      },
      "required": [
        "json_path",
        "description"
      ],
      "title": "FieldMetadata",
      "type": "object"
    },
    "IOMappingAgentMetadata": {
      "properties": {
        "input_fields": {
          "description": "an array of json paths representing fields to be used by io mapper in the mapping",
          "items": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "$ref": "#/$defs/FieldMetadata"
              }
            ]
          },
          "title": "Input Fields",
          "type": "array"
        },
        "output_fields": {
          "description": "an array of json paths representing firlds to be used by io mapper in the result",
          "items": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "$ref": "#/$defs/FieldMetadata"
              }
            ]
          },
          "title": "Output Fields",
          "type": "array"
        },
        "input_schema": {
          "anyOf": [
            {
              "additionalProperties": true,
              "type": "object"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "defines the schema for the input data",
          "title": "Input Schema"
        },
        "output_schema": {
          "anyOf": [
            {
              "additionalProperties": true,
              "type": "object"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "defines the schema for result of the mapping",
          "title": "Output Schema"
        },
        "output_description_prompt": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "A prompt structured using a Jinja template that will be used by the llm for a better mapping",
          "title": "Output Description Prompt"
        },
        "field_mapping": {
          "anyOf": [
            {
              "additionalProperties": {
                "type": "string"
              },
              "type": "object"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "A dictionary representing how the imperative mapping should be done where the keys are fields of the output object and values are JSONPath (strings)",
          "title": "Field Mapping"
        }
      },
      "required": [
        "input_fields",
        "output_fields"
      ],
      "title": "IOMappingAgentMetadata",
      "type": "object"
    }
  },
  "description": "This class exposes all\nThe IOMappingAgent class is designed for developers building sophisticated multi-agent software that require seamless integration and interaction between\nthe different agents and workflow steps.\nBy utilizing the methods provided, developers can construct complex workflows and softwares.\nThe IOMappingAgent class is intended to serve as a foundational component in applications requiring advanced IO mapping agents in multi-agent systems.",
  "properties": {
    "metadata": {
      "anyOf": [
        {
          "$ref": "#/$defs/IOMappingAgentMetadata"
        },
        {
          "type": "null"
        }
      ],
      "description": "Details about the fields to be used in the translation and about the output"
    },
    "llm": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Model to use for translation as LangChain description or model class.",
      "title": "Llm"
    }
  },
  "required": [
    "metadata"
  ],
  "title": "IOMappingAgent",
  "type": "object"
}

Fields:

Validators:

  • _validate_obj
Source code in .venv/lib/python3.13/site-packages/agntcy_iomapper/agent/agent_io_mapper.py 
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
class IOMappingAgent(BaseModel):
    """This class exposes all
    The IOMappingAgent class is designed for developers building sophisticated multi-agent software that require seamless integration and interaction between
    the different agents and workflow steps.
    By utilizing the methods provided, developers can construct complex workflows and softwares.
    The IOMappingAgent class is intended to serve as a foundational component in applications requiring advanced IO mapping agents in multi-agent systems.
    """

    metadata: Optional[IOMappingAgentMetadata] = Field(
        ...,
        description="Details about the fields to be used in the translation and about the output",
    )
    llm: Optional[Union[BaseChatModel, str]] = Field(
        None,
        description="Model to use for translation as LangChain description or model class.",
    )

    @model_validator(mode="after")
    def _validate_obj(self) -> Self:
        if not self.metadata:
            return self

        if not self.metadata.input_fields or len(self.metadata.input_fields) == 0:
            raise ValueError("input_fields not found in the metadata")
        # input fields must have a least one non empty string
        valid_input = [
            field
            for field in self.metadata.input_fields
            if (isinstance(field, str) and len(field.strip()) > 0)
            or (isinstance(field, FieldMetadata) and len(field.json_path.strip()) > 0)
        ]

        if not len(valid_input):
            raise ValueError("input_fields must have at least one field")
        else:
            self.metadata.input_fields = valid_input

        if not self.metadata.output_fields:
            raise ValueError("output_fields not found in the metadata")

        # output fields must have a least one non empty string
        valid_output = [
            field
            for field in self.metadata.output_fields
            if (isinstance(field, str) and len(field.strip()) > 0)
            or (isinstance(field, FieldMetadata) and len(field.json_path.strip()) > 0)
        ]

        if not len(valid_output):
            raise ValueError("output_fields must have at least one field")
        else:
            self.metadata.output_fields = valid_output

        return self

    def langgraph_node(self, data: Any, config: Optional[dict] = None) -> Runnable:
        """This method is used to add a language graph node to a langgraph multi-agent software.
        It leverages language models for IO mapping, ensuring efficient communication between agents.
        """

        input_type, output_type = get_io_types(data, self.metadata)

        data_to_be_mapped = extract_nested_fields(
            data, fields=self.metadata.input_fields
        )

        input = AgentIOMapperInput(
            input=ArgumentsDescription(
                json_schema=input_type,
            ),
            output=ArgumentsDescription(json_schema=output_type),
            data=data_to_be_mapped,
        )

        if not self.llm and config:
            configurable = config.get("configurable", None)
            if configurable is None:
                raise ValueError("llm instance not provided")

            llm = configurable.get("llm", None)

            if llm is None:
                raise ValueError("llm instance not provided")

            self.llm = llm

        if not self.llm:
            raise ValueError("llm instance not provided")

        iomapper_config = LangGraphIOMapperConfig(llm=self.llm)
        return LangGraphIOMapper(iomapper_config, input).as_runnable()

    def langgraph_imperative(
        self, data: Any, config: Optional[dict] = None
    ) -> Runnable:
        """
        Description: Similar to langgraph_node, this method adds a language graph node to a multi-agent software.
        However, it does not utilize a language model for IO mapping, offering an imperative approach to agent integration.
        """

        input_type, output_type = get_io_types(data, self.metadata)

        data_to_be_mapped = extract_nested_fields(
            data, fields=self.metadata.input_fields
        )

        input = ImperativeIOMapperInput(
            input=ArgumentsDescription(
                json_schema=input_type,
            ),
            output=ArgumentsDescription(json_schema=output_type),
            data=data_to_be_mapped,
        )

        if not self.metadata.field_mapping:
            raise ValueError(
                "In order to use imperative mapping field_mapping must be provided in the metadata"
            )

        imperative_io_mapper = ImperativeIOMapper(
            input=input, field_mapping=self.metadata.field_mapping
        )
        return imperative_io_mapper.as_runnable()

    @staticmethod
    def as_worfklow_step(workflow: Workflow) -> Callable:
        """This static method allows for the addition of a step to a LlamaIndex workflow.
        It integrates seamlessly into workflows, enabling structured progression and task execution.
        """
        io_mapper_step = LLamaIndexIOMapper.llamaindex_mapper(workflow)
        return io_mapper_step

    @staticmethod
    def as_workflow_agent(
        mapping_metadata: IOMappingAgentMetadata,
        llm: BaseLLM,
        name: str,
        description: str,
        can_handoff_to: Optional[List[str]] = None,
        tools: Optional[List[Union[BaseTool, Callable]]] = [],
    ):
        """This static method returns an instance of an agent that can be integrated into a Multi AgentWorkflow.
        It provides robust IO mapping capabilities essential for complex multi agent workflow interactions.
        """
        return LLamaIndexIOMapper(
            mapping_metadata=mapping_metadata,
            llm=llm,
            tools=tools,
            name=name,
            description=description,
            can_handoff_to=can_handoff_to,
        )

llm = None pydantic-field

Model to use for translation as LangChain description or model class.

metadata pydantic-field

Details about the fields to be used in the translation and about the output

as_worfklow_step(workflow) staticmethod

This static method allows for the addition of a step to a LlamaIndex workflow. It integrates seamlessly into workflows, enabling structured progression and task execution.

Source code in .venv/lib/python3.13/site-packages/agntcy_iomapper/agent/agent_io_mapper.py 
160
161
162
163
164
165
166
@staticmethod
def as_worfklow_step(workflow: Workflow) -> Callable:
    """This static method allows for the addition of a step to a LlamaIndex workflow.
    It integrates seamlessly into workflows, enabling structured progression and task execution.
    """
    io_mapper_step = LLamaIndexIOMapper.llamaindex_mapper(workflow)
    return io_mapper_step

as_workflow_agent(mapping_metadata, llm, name, description, can_handoff_to=None, tools=[]) staticmethod

This static method returns an instance of an agent that can be integrated into a Multi AgentWorkflow. It provides robust IO mapping capabilities essential for complex multi agent workflow interactions.

Source code in .venv/lib/python3.13/site-packages/agntcy_iomapper/agent/agent_io_mapper.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
@staticmethod
def as_workflow_agent(
    mapping_metadata: IOMappingAgentMetadata,
    llm: BaseLLM,
    name: str,
    description: str,
    can_handoff_to: Optional[List[str]] = None,
    tools: Optional[List[Union[BaseTool, Callable]]] = [],
):
    """This static method returns an instance of an agent that can be integrated into a Multi AgentWorkflow.
    It provides robust IO mapping capabilities essential for complex multi agent workflow interactions.
    """
    return LLamaIndexIOMapper(
        mapping_metadata=mapping_metadata,
        llm=llm,
        tools=tools,
        name=name,
        description=description,
        can_handoff_to=can_handoff_to,
    )

langgraph_imperative(data, config=None)

Description: Similar to langgraph_node, this method adds a language graph node to a multi-agent software. However, it does not utilize a language model for IO mapping, offering an imperative approach to agent integration.

Source code in .venv/lib/python3.13/site-packages/agntcy_iomapper/agent/agent_io_mapper.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def langgraph_imperative(
    self, data: Any, config: Optional[dict] = None
) -> Runnable:
    """
    Description: Similar to langgraph_node, this method adds a language graph node to a multi-agent software.
    However, it does not utilize a language model for IO mapping, offering an imperative approach to agent integration.
    """

    input_type, output_type = get_io_types(data, self.metadata)

    data_to_be_mapped = extract_nested_fields(
        data, fields=self.metadata.input_fields
    )

    input = ImperativeIOMapperInput(
        input=ArgumentsDescription(
            json_schema=input_type,
        ),
        output=ArgumentsDescription(json_schema=output_type),
        data=data_to_be_mapped,
    )

    if not self.metadata.field_mapping:
        raise ValueError(
            "In order to use imperative mapping field_mapping must be provided in the metadata"
        )

    imperative_io_mapper = ImperativeIOMapper(
        input=input, field_mapping=self.metadata.field_mapping
    )
    return imperative_io_mapper.as_runnable()

langgraph_node(data, config=None)

This method is used to add a language graph node to a langgraph multi-agent software. It leverages language models for IO mapping, ensuring efficient communication between agents.

Source code in .venv/lib/python3.13/site-packages/agntcy_iomapper/agent/agent_io_mapper.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def langgraph_node(self, data: Any, config: Optional[dict] = None) -> Runnable:
    """This method is used to add a language graph node to a langgraph multi-agent software.
    It leverages language models for IO mapping, ensuring efficient communication between agents.
    """

    input_type, output_type = get_io_types(data, self.metadata)

    data_to_be_mapped = extract_nested_fields(
        data, fields=self.metadata.input_fields
    )

    input = AgentIOMapperInput(
        input=ArgumentsDescription(
            json_schema=input_type,
        ),
        output=ArgumentsDescription(json_schema=output_type),
        data=data_to_be_mapped,
    )

    if not self.llm and config:
        configurable = config.get("configurable", None)
        if configurable is None:
            raise ValueError("llm instance not provided")

        llm = configurable.get("llm", None)

        if llm is None:
            raise ValueError("llm instance not provided")

        self.llm = llm

    if not self.llm:
        raise ValueError("llm instance not provided")

    iomapper_config = LangGraphIOMapperConfig(llm=self.llm)
    return LangGraphIOMapper(iomapper_config, input).as_runnable()

LangGraph Example 1

This example involves a multi-agent software system designed to process a create engagement campaign and share within an organization. It interacts with an agent specialized in creating campaigns, another agent specialized in identifying suitable users. The information is then relayed to an IO mapper, which converts the list of users and the campaign details to present statistics about the campaign.

Define an agent io mapper metadata

metadata = IOMappingAgentMetadata(
    input_fields=["selected_users", "campaign_details.name"],
    output_fields=["stats.status"],
)

The above instruction directs the IO mapper agent to utilize the selected_users and name from the campaign_details field and map them to the stats.status. No further information is needed since the type information can be derived from the input data which is a pydantic model.

Tip

Both input_fields and output_fields can also be sourced with a list composed of str and/or instances of FieldMetadata as the bellow example shows

metadata = IOMappingAgentMetadata(
    input_fields=[
        FieldMetadata(
            json_path="selected_users", description="A list of users to be targeted"
        ),
        FieldMetadata(
            json_path="campaign_details.name",
            description="The name that can be used by the campaign",
            examples=["Campaign A"]
        ),
    ],
    output_fields=["stats"],
)

Define an Instance of the Agent

mapping_agent = IOMappingAgent(metadata=metadata, llm=llm)

Add the node to the LangGraph graph

workflow.add_node(
    "io_mapping",
    mapping_agent.langgraph_node,
)

Add the Edge

With the edge added, you can run the your LangGraph graph.

workflow.add_edge("create_communication", "io_mapping")
workflow.add_edge("io_mapping", "send_communication")

LangGraph Example 2

This example involves a multi-agent software system designed to process a list of ingredients. It interacts with an agent specialized in recipe books to identify feasible recipes based on the provided ingredients. The information is then relayed to an IO mapper, which converts it into a format suitable for display to the user.

Define an Agent IO Mapper Metadata

metadata = IOMappingAgentMetadata(
    input_fields=["documents.0.page_content"],
    output_fields=["recipe"],
    input_schema=TypeAdapter(GraphState).json_schema(),
    output_schema={
        "type": "object",
        "properties": {
            "title": {"type": "string"},
            "ingredients": {"type": "array", "items": {"type": "string"}},
            "instructions": {"type": "string"},
        },
        "required": ["title", "ingredients, instructions"],
    },
)

Define an Instance of the Agent

mapping_agent = IOMappingAgent(metadata=metadata, llm=llm)

Add the node to the LangGraph graph

graph.add_node(
    "recipe_io_mapper",
    mapping_agent.langgraph_node,
)

Add the Edge

With the edge added, you can run the your LangGraph graph.

graph.add_edge("recipe_expert", "recipe_io_mapper")

LlamaIndex

We support both LlamaIndex Workflow and the new AgentWorkflow multi agent software

Entities

IOMappingInputEvent

... agntcy_iomapper.IOMappingInputEvent

IOMappingOutputEvent

... agntcy_iomapper.IOMappingOutputEvent

Example of usage in a LlamaIndex workflow

In this example we recreate the campaign workflow using LlamaIndex workflow <https://docs.llamaindex.ai/en/stable/module_guides/workflow/>_

Begin by importing the neccessary object

from agntcy_iomapper import IOMappingAgent, IOMappingAgentMetadata

Define the workflow

class CampaignWorkflow(Workflow):
    @step
    async def prompt_step(self, ctx: Context, ev: StartEvent) -> PickUsersEvent:
        await ctx.set("llm", ev.get("llm"))
        return PickUsersEvent(prompt=ev.get("prompt"))

    @step
    async def pick_users_step(
        self, ctx: Context, ev: PickUsersEvent
    ) -> CreateCampaignEvent:
        return CreateCampaignEvent(list_users=users)

    # The step that will trigger IO mapping
    @step
    async def create_campaign(
        self, ctx: Context, ev: CreateCampaignEvent
    ) -> IOMappingInputEvent:
        prompt = f"""
        You are a campaign builder for company XYZ. Given a list of selected users and a user prompt, create an engaging campaign.
        Return the campaign details as a JSON object with the following structure:
        {{
            "name": "Campaign Name",
            "content": "Campaign Content",
            "is_urgent": yes/no
        }}
        Selected Users: {ev.list_users}
        User Prompt: Create a campaign for all users
        """
        parser = PydanticOutputParser(output_cls=Campaign)
        llm = await ctx.get("llm", default=None)

        llm_response = llm.complete(prompt)
        try:
            campaign_details = parser.parse(str(llm_response))
            metadata = IOMappingAgentMetadata(
                input_fields=["selected_users", "campaign_details.name"],
                output_fields=["stats"],
            )
            config = LLamaIndexIOMapperConfig(llm=llm)

            io_mapping_input_event = IOMappingInputEvent(
                metadata=metadata,
                config=config,
                data=OverallState(
                    campaign_details=campaign_details,
                    selected_users=ev.list_users,
                ),
            )
            return io_mapping_input_event
        except Exception as e:
            print(f"Error parsing campaign details: {e}")
            return StopEvent(result=f"{e}")

    @step
    async def after_translation(self, evt: IOMappingOutputEvent) -> StopEvent:
        return StopEvent(result="Done")

Tip

The highlighted lines shows how the io mapper can be triggered

Add The IO mapper step

w = CampaignWorkflow()
IOMappingAgent.as_worfklow_step(workflow=w)

Example of usage in a LlamaIndex AgentWorkflow

In this example we recreate the recipe workflow using LlamaIndex AgentWorkflow <https://docs.llamaindex.ai/en/stable/module_guides/workflow/>_

Import the necessary objects

from agntcy_iomapper import FieldMetadata, IOMappingAgent, IOMappingAgentMetadata

Define an instance of the IOMappingAgentMetadata

mapping_metadata = IOMappingAgentMetadata(
        input_fields=["documents.0.text"],
        output_fields=[
            FieldMetadata(
                json_path="recipe",
                description="this is a recipe for the ingredients you've provided",
            )
        ],
        input_schema=TypeAdapter(GraphState).json_schema(),
        output_schema={
            "type": "object",
            "properties": {
                "title": {"type": "string"},
                "ingredients": {"type": "array", "items": {"type": "string"}},
                "instructions": {"type": "string"},
            },
            "required": ["title", "ingredients, instructions"],
        },
    )

Finally define the IOMappingAgent and add it to the AgentWorkflow.

Important to note that a tool is passed, to instruct the io mapper where to go next in the flow.

io_mapping_agent = IOMappingAgent.as_workflow_agent(
    mapping_metadata=mapping_metadata,
    llm=llm,
    name="IOMapperAgent",
    description="Useful for mapping a recipe document into recipe object",
    can_handoff_to=["Formatter_Agent"],
    tools=[got_to_format],
)


io_mapping_agent = IOMappingAgent.as_workflow_agent(
    mapping_metadata=mapping_metadata,
    llm=llm,
    name="IOMapperAgent",
    description="Useful for mapping a recipe document into recipe object",
    can_handoff_to=["Formatter_Agent"],
    tools=[got_to_format],
)

Use Examples

  1. Install:
  2. cmake <https://cmake.org/>_

  3. pip <https://pip.pypa.io/en/stable/installation/>_

  4. From the examples folder run the desired make command, for example:

make make run_lg_eg_py

Contributing

Contributions are what make the open source community such an amazing place to learn, inspire, and create. Any contributions you make are greatly appreciated. For detailed contributing guidelines, please see CONTRIBUTING.md <https://github.com/agntcy/acp-sdk/blob/main/docs/CONTRIBUTING.md>_

Copyright Notice and License <https://github.com/agntcy/acp-sdk/blob/main/LICENSE>_

Copyright (c) 2025 Cisco and/or its affiliates.

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

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.