Repository: deepset-ai/haystack

Issue:
Agent component doesn't have an output called `messages`
**Describe the bug**
According to [documentation](https://docs.haystack.deepset.ai/docs/agent) `Agent` component should have an output called `messages`.  But it seems that, using `Agent` component in a pipeline and trying to connect its output `messages` to another component, `messages` output doesn't exist.

**To Reproduce**
```python
from haystack import Pipeline
from haystack.components.generators.chat import AzureOpenAIChatGenerator
from haystack.components.joiners import ListJoiner
from haystack.dataclasses import ChatMessage
from typing import List

pipeline = Pipeline()

from haystack.components.websearch import SerperDevWebSearch
from haystack.tools import ComponentTool
from haystack.components.agents import Agent

search_tool = ComponentTool(component=SerperDevWebSearch(top_k=3))
agent = Agent(
    chat_generator=AzureOpenAIChatGenerator(tools=[search_tool]),
    system_prompt="""You're a helpful agent. When asked about current information like weather, news, or facts, 
                     use the web_search tool to find the information and then summarize the findings.
                     When you get web search results, extract the relevant information and present it in a clear, 
                     concise manner.""",
    tools=[search_tool],
)

pipeline.add_component("agent", agent)
pipeline.add_component("joiner", ListJoiner(List[ChatMessage]))
pipeline.connect("agent.messages", "joiner.values")

res = pipeline.run(data={
   "messages": [ChatMessage.from_user("How is the weather in Berlin?")],
})

print(res)
```

**Error message**
```python
pipeline.connect("agent.messages", "joiner.values")
  File "***/haystack/core/pipeline/base.py", line 457, in connect
    raise PipelineConnectError(
haystack.core.errors.PipelineConnectError: 'agent.messages does not exist. Output connections of agent are: 
```

**Additional context**
Looking at [code](https://github.com/deepset-ai/haystack/blob/main/haystack/components/agents/agent.py) I don't see output is set.


**System:**
 - Haystack version (commit or version number): 2.12

Hints from issue comments:
@sjrl: Hey @danielescaramuzzi thanks for raising! I opened a PR with the fix which I'll finish up tomorrow.@sjrl: Hey @danielescaramuzzi this is works now in haystack==2.12.1

Solution patch:
diff --git a/haystack/components/agents/agent.py b/haystack/components/agents/agent.py
index 16902e79f8..b301032c0f 100644
--- a/haystack/components/agents/agent.py
+++ b/haystack/components/agents/agent.py
@@ -3,6 +3,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import inspect
+from copy import deepcopy
 from typing import Any, Dict, List, Optional
 
 from haystack import component, default_from_dict, default_to_dict, logging
@@ -10,6 +11,7 @@
 from haystack.components.tools import ToolInvoker
 from haystack.dataclasses import ChatMessage
 from haystack.dataclasses.state import State, _schema_from_dict, _schema_to_dict, _validate_schema
+from haystack.dataclasses.state_utils import merge_lists
 from haystack.dataclasses.streaming_chunk import SyncStreamingCallbackT
 from haystack.tools import Tool, deserialize_tools_or_toolset_inplace
 from haystack.utils.callable_serialization import deserialize_callable, serialize_callable
@@ -97,9 +99,16 @@ def __init__(
                 "Ensure that each exit condition corresponds to either 'text' or a valid tool name."
             )
 
+        # Validate state schema if provided
         if state_schema is not None:
             _validate_schema(state_schema)
-        self.state_schema = state_schema or {}
+        self._state_schema = state_schema or {}
+
+        # Initialize state schema
+        resolved_state_schema = deepcopy(self._state_schema)
+        if resolved_state_schema.get("messages") is None:
+            resolved_state_schema["messages"] = {"type": List[ChatMessage], "handler": merge_lists}
+        self.state_schema = resolved_state_schema
 
         self.chat_generator = chat_generator
         self.tools = tools or []
@@ -111,8 +120,11 @@ def __init__(
 
         output_types = {}
         for param, config in self.state_schema.items():
-            component.set_input_type(self, name=param, type=config["type"], default=None)
             output_types[param] = config["type"]
+            # Skip setting input types for parameters that are already in the run method
+            if param in ["messages", "streaming_callback"]:
+                continue
+            component.set_input_type(self, name=param, type=config["type"], default=None)
         component.set_output_types(self, **output_types)
 
         self._tool_invoker = ToolInvoker(tools=self.tools, raise_on_failure=self.raise_on_tool_invocation_failure)
@@ -144,7 +156,8 @@ def to_dict(self) -> Dict[str, Any]:
             tools=[t.to_dict() for t in self.tools],
             system_prompt=self.system_prompt,
             exit_conditions=self.exit_conditions,
-            state_schema=_schema_to_dict(self.state_schema),
+            # We serialize the original state schema, not the resolved one to reflect the original user input
+            state_schema=_schema_to_dict(self._state_schema),
             max_agent_steps=self.max_agent_steps,
             raise_on_tool_invocation_failure=self.raise_on_tool_invocation_failure,
             streaming_callback=streaming_callback,
diff --git a/haystack/dataclasses/state.py b/haystack/dataclasses/state.py
index 02fc3c63e1..14b884820d 100644
--- a/haystack/dataclasses/state.py
+++ b/haystack/dataclasses/state.py
@@ -2,6 +2,7 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
+from copy import deepcopy
 from typing import Any, Callable, Dict, List, Optional
 
 from haystack.dataclasses import ChatMessage
@@ -93,13 +94,13 @@ def __init__(self, schema: Dict[str, Any], data: Optional[Dict[str, Any]] = None
         :param data: Optional dictionary of initial data to populate the state
         """
         _validate_schema(schema)
-        self.schema = schema
+        self.schema = deepcopy(schema)
         if self.schema.get("messages") is None:
             self.schema["messages"] = {"type": List[ChatMessage], "handler": merge_lists}
         self._data = data or {}
 
         # Set default handlers if not provided in schema
-        for definition in schema.values():
+        for definition in self.schema.values():
             # Skip if handler is already defined and not None
             if definition.get("handler") is not None:
                 continue
@@ -117,7 +118,7 @@ def get(self, key: str, default: Any = None) -> Any:
         :param default: Value to return if key is not found
         :returns: Value associated with key or default if not found
         """
-        return self._data.get(key, default)
+        return deepcopy(self._data.get(key, default))
 
     def set(self, key: str, value: Any, handler_override: Optional[Callable[[Any, Any], Any]] = None) -> None:
         """
diff --git a/releasenotes/notes/fix-agent-output-messages-550404a95199bb67.yaml b/releasenotes/notes/fix-agent-output-messages-550404a95199bb67.yaml
new file mode 100644
index 0000000000..452772b6d6
--- /dev/null
+++ b/releasenotes/notes/fix-agent-output-messages-550404a95199bb67.yaml
@@ -0,0 +1,5 @@
+---
+fixes:
+  - |
+    In Agent we make sure state_schema is always initialized to have 'messages'. Previously this was only happening at run time which is why pipeline.connect failed because output types are set at init time.
+    Now the Agent correctly sets everything in state_schema (including messages by default) at init time.


Test patch:
diff --git a/test/components/agents/test_agent.py b/test/components/agents/test_agent.py
index 92d58dab3a..b7e1814209 100644
--- a/test/components/agents/test_agent.py
+++ b/test/components/agents/test_agent.py
@@ -16,10 +16,12 @@
 from haystack.components.builders.prompt_builder import PromptBuilder
 from haystack.components.generators.chat.openai import OpenAIChatGenerator
 from haystack.components.generators.chat.types import ChatGenerator
+from haystack.core.component.types import OutputSocket
 from haystack.dataclasses import ChatMessage, ToolCall
 from haystack.dataclasses.streaming_chunk import StreamingChunk
 from haystack.tools import Tool, ComponentTool
 from haystack.utils import serialize_callable, Secret
+from haystack.dataclasses.state_utils import merge_lists
 
 
 def streaming_callback_for_serde(chunk: StreamingChunk):
@@ -103,6 +105,166 @@ def run(self, messages: List[ChatMessage]) -> Dict[str, Any]:
 
 
 class TestAgent:
+    def test_output_types(self, weather_tool, component_tool, monkeypatch):
+        monkeypatch.setenv("OPENAI_API_KEY", "fake-key")
+        chat_generator = OpenAIChatGenerator()
+        agent = Agent(chat_generator=chat_generator, tools=[weather_tool, component_tool])
+        assert agent.__haystack_output__._sockets_dict == {
+            "messages": OutputSocket(name="messages", type=List[ChatMessage], receivers=[])
+        }
+
+    def test_to_dict(self, weather_tool, component_tool, monkeypatch):
+        monkeypatch.setenv("OPENAI_API_KEY", "fake-key")
+        generator = OpenAIChatGenerator()
+        agent = Agent(
+            chat_generator=generator,
+            tools=[weather_tool, component_tool],
+            exit_conditions=["text", "weather_tool"],
+            state_schema={"foo": {"type": str}},
+        )
+        serialized_agent = agent.to_dict()
+        assert serialized_agent == {
+            "type": "haystack.components.agents.agent.Agent",
+            "init_parameters": {
+                "chat_generator": {
+                    "type": "haystack.components.generators.chat.openai.OpenAIChatGenerator",
+                    "init_parameters": {
+                        "model": "gpt-4o-mini",
+                        "streaming_callback": None,
+                        "api_base_url": None,
+                        "organization": None,
+                        "generation_kwargs": {},
+                        "api_key": {"type": "env_var", "env_vars": ["OPENAI_API_KEY"], "strict": True},
+                        "timeout": None,
+                        "max_retries": None,
+                        "tools": None,
+                        "tools_strict": False,
+                    },
+                },
+                "tools": [
+                    {
+                        "type": "haystack.tools.tool.Tool",
+                        "data": {
+                            "name": "weather_tool",
+                            "description": "Provides weather information for a given location.",
+                            "parameters": {
+                                "type": "object",
+                                "properties": {"location": {"type": "string"}},
+                                "required": ["location"],
+                            },
+                            "function": "test_agent.weather_function",
+                            "outputs_to_string": None,
+                            "inputs_from_state": None,
+                            "outputs_to_state": None,
+                        },
+                    },
+                    {
+                        "type": "haystack.tools.component_tool.ComponentTool",
+                        "data": {
+                            "component": {
+                                "type": "haystack.components.builders.prompt_builder.PromptBuilder",
+                                "init_parameters": {
+                                    "template": "{{parrot}}",
+                                    "variables": None,
+                                    "required_variables": None,
+                                },
+                            },
+                            "name": "parrot",
+                            "description": "This is a parrot.",
+                            "parameters": None,
+                            "outputs_to_string": None,
+                            "inputs_from_state": None,
+                            "outputs_to_state": None,
+                        },
+                    },
+                ],
+                "system_prompt": None,
+                "exit_conditions": ["text", "weather_tool"],
+                "state_schema": {"foo": {"type": "str"}},
+                "max_agent_steps": 100,
+                "raise_on_tool_invocation_failure": False,
+                "streaming_callback": None,
+            },
+        }
+
+    def test_from_dict(self, weather_tool, component_tool, monkeypatch):
+        monkeypatch.setenv("OPENAI_API_KEY", "fake-key")
+        data = {
+            "type": "haystack.components.agents.agent.Agent",
+            "init_parameters": {
+                "chat_generator": {
+                    "type": "haystack.components.generators.chat.openai.OpenAIChatGenerator",
+                    "init_parameters": {
+                        "model": "gpt-4o-mini",
+                        "streaming_callback": None,
+                        "api_base_url": None,
+                        "organization": None,
+                        "generation_kwargs": {},
+                        "api_key": {"type": "env_var", "env_vars": ["OPENAI_API_KEY"], "strict": True},
+                        "timeout": None,
+                        "max_retries": None,
+                        "tools": None,
+                        "tools_strict": False,
+                    },
+                },
+                "tools": [
+                    {
+                        "type": "haystack.tools.tool.Tool",
+                        "data": {
+                            "name": "weather_tool",
+                            "description": "Provides weather information for a given location.",
+                            "parameters": {
+                                "type": "object",
+                                "properties": {"location": {"type": "string"}},
+                                "required": ["location"],
+                            },
+                            "function": "test_agent.weather_function",
+                            "outputs_to_string": None,
+                            "inputs_from_state": None,
+                            "outputs_to_state": None,
+                        },
+                    },
+                    {
+                        "type": "haystack.tools.component_tool.ComponentTool",
+                        "data": {
+                            "component": {
+                                "type": "haystack.components.builders.prompt_builder.PromptBuilder",
+                                "init_parameters": {
+                                    "template": "{{parrot}}",
+                                    "variables": None,
+                                    "required_variables": None,
+                                },
+                            },
+                            "name": "parrot",
+                            "description": "This is a parrot.",
+                            "parameters": None,
+                            "outputs_to_string": None,
+                            "inputs_from_state": None,
+                            "outputs_to_state": None,
+                        },
+                    },
+                ],
+                "system_prompt": None,
+                "exit_conditions": ["text", "weather_tool"],
+                "state_schema": {"foo": {"type": "str"}},
+                "max_agent_steps": 100,
+                "raise_on_tool_invocation_failure": False,
+                "streaming_callback": None,
+            },
+        }
+        agent = Agent.from_dict(data)
+        assert isinstance(agent, Agent)
+        assert isinstance(agent.chat_generator, OpenAIChatGenerator)
+        assert agent.chat_generator.model == "gpt-4o-mini"
+        assert agent.chat_generator.api_key == Secret.from_env_var("OPENAI_API_KEY")
+        assert agent.tools[0].function is weather_function
+        assert isinstance(agent.tools[1]._component, PromptBuilder)
+        assert agent.exit_conditions == ["text", "weather_tool"]
+        assert agent.state_schema == {
+            "foo": {"type": str},
+            "messages": {"handler": merge_lists, "type": List[ChatMessage]},
+        }
+
     def test_serde(self, weather_tool, component_tool, monkeypatch):
         monkeypatch.setenv("FAKE_OPENAI_KEY", "fake-key")
         generator = OpenAIChatGenerator(api_key=Secret.from_env_var("FAKE_OPENAI_KEY"))
@@ -122,7 +284,7 @@ def test_serde(self, weather_tool, component_tool, monkeypatch):
             init_parameters["chat_generator"]["type"]
             == "haystack.components.generators.chat.openai.OpenAIChatGenerator"
         )
-        assert init_parameters["streaming_callback"] == None
+        assert init_parameters["streaming_callback"] is None
         assert init_parameters["tools"][0]["data"]["function"] == serialize_callable(weather_function)
         assert (
             init_parameters["tools"][1]["data"]["component"]["type"]
@@ -137,7 +299,10 @@ def test_serde(self, weather_tool, component_tool, monkeypatch):
         assert deserialized_agent.tools[0].function is weather_function
         assert isinstance(deserialized_agent.tools[1]._component, PromptBuilder)
         assert deserialized_agent.exit_conditions == ["text", "weather_tool"]
-        assert deserialized_agent.state_schema == {"foo": {"type": str}}
+        assert deserialized_agent.state_schema == {
+            "foo": {"type": str},
+            "messages": {"handler": merge_lists, "type": List[ChatMessage]},
+        }
 
     def test_serde_with_streaming_callback(self, weather_tool, component_tool, monkeypatch):
         monkeypatch.setenv("FAKE_OPENAI_KEY", "fake-key")
diff --git a/test/dataclasses/test_state.py b/test/dataclasses/test_state.py
index 2a9d23aecd..977d502510 100644
--- a/test/dataclasses/test_state.py
+++ b/test/dataclasses/test_state.py
@@ -2,7 +2,7 @@
 from typing import List, Dict
 
 from haystack.dataclasses import ChatMessage
-from haystack.dataclasses.state import State, _validate_schema
+from haystack.dataclasses.state import State, _validate_schema, _schema_to_dict, _schema_from_dict
 from haystack.dataclasses.state_utils import merge_lists
 
 
@@ -11,16 +11,15 @@ def basic_schema():
     return {"numbers": {"type": list}, "metadata": {"type": dict}, "name": {"type": str}}
 
 
+def numbers_handler(current, new):
+    if current is None:
+        return sorted(set(new))
+    return sorted(set(current + new))
+
+
 @pytest.fixture
 def complex_schema():
-    return {
-        "numbers": {
-            "type": list,
-            "handler": lambda current, new: sorted(set(current + new)) if current else sorted(set(new)),
-        },
-        "metadata": {"type": dict},
-        "name": {"type": str},
-    }
+    return {"numbers": {"type": list, "handler": numbers_handler}, "metadata": {"type": dict}, "name": {"type": str}}
 
 
 def test_validate_schema_valid(basic_schema):
@@ -147,3 +146,44 @@ def test_state_nested_structures():
 
     expected = {"a": [1, 2], "b": [3, 4, 5, 6], "c": [7, 8]}
     assert state.get("complex") == expected
+
+
+def test_schema_to_dict(basic_schema):
+    expected_dict = {"numbers": {"type": "list"}, "metadata": {"type": "dict"}, "name": {"type": "str"}}
+    result = _schema_to_dict(basic_schema)
+    assert result == expected_dict
+
+
+def test_schema_to_dict_with_handlers(complex_schema):
+    expected_dict = {
+        "numbers": {"type": "list", "handler": "test_state.numbers_handler"},
+        "metadata": {"type": "dict"},
+        "name": {"type": "str"},
+    }
+    result = _schema_to_dict(complex_schema)
+    assert result == expected_dict
+
+
+def test_schema_from_dict(basic_schema):
+    schema_dict = {"numbers": {"type": "list"}, "metadata": {"type": "dict"}, "name": {"type": "str"}}
+    result = _schema_from_dict(schema_dict)
+    assert result == basic_schema
+
+
+def test_schema_from_dict_with_handlers(complex_schema):
+    schema_dict = {
+        "numbers": {"type": "list", "handler": "test_state.numbers_handler"},
+        "metadata": {"type": "dict"},
+        "name": {"type": "str"},
+    }
+    result = _schema_from_dict(schema_dict)
+    assert result == complex_schema
+
+
+def test_state_mutability():
+    state = State({"my_list": {"type": list}}, {"my_list": [1, 2]})
+
+    my_list = state.get("my_list")
+    my_list.append(3)
+
+    assert state.get("my_list") == [1, 2]