agntcy_acp.langgraph package

Submodules

agntcy_acp.langgraph.acp_node module

class agntcy_acp.langgraph.acp_node.ACPNode(name, agent_id, client_config, input_path, input_type, output_path, output_type, config_path=None, config_type=None, auth_header=None, use_threads=False)

Bases: object

This class represents a Langgraph Node that holds a remote connection to an ACP Agent It can be instantiated and added to any langgraph graph.

my_node = ACPNode(…) sg = StateGraph(GraphState) sg.add_node(my_node)

async ainvoke(state, config=None, **kwargs)

Return type:

Any

async astream(input, config=None, *, stream_mode=None, **kwargs)

Stream graph steps for a single input.

Return type:

AsyncIterator[Union[Dict[str, Any], Any]]

get_name()

invoke(state, config=None, **kwargs)

Return type:

Any

stream(input, config=None, *, stream_mode=None, **kwargs)

Stream graph steps for a single input.

Return type:

Iterator[Union[Dict[str, Any], Any]]

agntcy_acp.langgraph.api_bridge module

class agntcy_acp.langgraph.api_bridge.APIBridgeAgentNode(name, hostname, service_name, input_path, output_path, service_api_key, input_type=<class 'agntcy_acp.langgraph.api_bridge.APIBridgeInput'>, output_type=<class 'agntcy_acp.langgraph.api_bridge.APIBridgeOutput'>, apikey=None)

Bases: ACPNode

An ACP node that enables using remotely the API bridge agent in a LangGraph multi agent software

async ainvoke(state, config)

Return type:

Any

invoke(state, config)

Return type:

Any

class agntcy_acp.langgraph.api_bridge.APIBridgeInput(**data)

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

query: str

class agntcy_acp.langgraph.api_bridge.APIBridgeOutput(**data)

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

result: str

agntcy_acp.langgraph.io_mapper module

agntcy_acp.langgraph.io_mapper.add_io_mapped_conditional_edge(g, start, path, iomapper_config_map, llm)

Adds a conditional I/O-mapped edge to a LangGraph StateGraph.

Return type:

dict

Parameters:

g: The LangGraph StateGraph to which the conditional edge will be added. start: The starting node of the edge, which can be specified either as a string identifier or as an instance of an ACPNode. path: The conditional path that determines the conditions under which the edge will be traversed. The type and structure of ‘path’ should be specified based on its use case. iomapper_config_map: A dictionary containing metadata that the IO mapper agent requires for data translation. This map is used to configure the agent based on different conditions. llm: An instance of llm model

Returns:

None: This function modifies the graph in place by adding the specified conditional edge.

agntcy_acp.langgraph.io_mapper.add_io_mapped_edge(g, start, end, iomapper_config, llm)

Adds an I/O-mapped edge to a LangGraph StateGraph.

Return type:

IOMappingAgent

Parameters:

g: The LangGraph StateGraph to which the edge will be added. start: The starting node of the edge, which can be specified either as a string identifier or as an instance of an ACPNode. end: The ending node of the edge, which can be specified either as a string identifier or as an instance of an ACPNode. iomapper_config: A dictionary containing all the metadata necessary for the IO mapper agent to perform data translation. Defaults to an empty dictionary. llm: An instance of llm model

Returns:

None: This function modifies the graph in place by adding the specified edge.

Module contents