`PipelineConnectError` caused by optional `None` input mis‑inference

[Pitrified]

Describe the bug

When a component’s run method accepts an optional list[str] | None, Haystack incorrectly infers its input type as list[list[str]] during pipeline connection. A similar component requiring a mandatory list[str] works fine.

Error message

PipelineConnectError: Cannot connect 'producer.words' with 'consumer_opt.words': 
their declared input and output types do not match.
'producer':
  - words: list[str]
'consumer_opt':
  - words: list[list[str]] (available)

Expected behavior

Since StringConsumerOptional explicitly declares words: list[str] | None, Haystack should allow the connection from a list[str] output-just as it does for the mandatory version-without inferring a nested list.

Minimal Reproduction Code

from typing import Any
from haystack import Pipeline, component

# Producer: outputs a plain list of strings
@component
class StringProducer:
    @component.output_types(words=list[str])
    def run(self) -> dict[str, Any]:
        return {"words": ["apple", "banana", "cherry"]}

# Consumer (optional): accepts list[str] | None
@component
class StringConsumerOptional:
    @component.output_types(count=int)
    def run(self, words: list[str] | None = None) -> dict[str, Any]:
        return {"count": len(words or [])}

pipeline = Pipeline()
pipeline.add_component("producer", StringProducer())
pipeline.add_component("consumer_opt", StringConsumerOptional())

# fails:
pipeline.connect("producer.words", "consumer_opt.words")

Additional context

Connection validation can be disabled (via Pipeline(..., connection_type_validation=False)), which may bypass this error. But this only masks the problem rather than preserving accurate type inference when optional inputs are used.

Please note that in the issues linked below the problem was Optional[str] -> str, which

1. is a different case: here, the issue seems with list[str] | None being misinterpreted as list[list[str]].
2. seems like a more dangerous case, as it can lead to runtime errors if the undeclared optional input is not handled correctly. in this case, the receiver can handle the None case gracefully and is declared to do so.

Please note that while this is a minimal example, there are valid scenarios where a component might need to accept an optional input of type list[str] | None.
We need a component where some of the inputs are optional. The component is used several times in the pipeline, and only in some cases the connection to that optional input is made.
In the actual code, there is at least one input that is always connected, so the component can still run successfully.

To Reproduce

Run the snippet above

FAQ Check

• Have you had a look at our new FAQ page?

System:

• OS: *Ubuntu 22.04
• GPU/CPU:
• Haystack version: 2.11.2
• Python: 3.13.1
• DocumentStore: (not applicable)
• Reader: (not applicable)
• Retriever: (not applicable)

Related Links

https://haystack.deepset.ai/release-notes/2.11.0
https://docs.haystack.deepset.ai/reference/pipeline-api#pipeline__init__
#8862
#8866
#8874

[sjrl]

@Pitrified thanks for opening the issue!

To be upfront we don't properly support the python native types in our type detection and checking. We plan on working on this but for the time being to unblock you I recommend making the following change (ie use types from the typing library so List, Optional

from typing import Any
from haystack import Pipeline, component
from typing import List, Optional

# Producer: outputs a plain list of strings
@component
class StringProducer:
    @component.output_types(words=List[str])
    def run(self) -> dict[str, Any]:
        return {"words": ["apple", "banana", "cherry"]}

# Consumer (optional): accepts list[str] | None
@component
class StringConsumerOptional:
    @component.output_types(count=int)
    def run(self, words: Optional[List[str]] = None) -> dict[str, Any]:
        return {"count": len(words or [])}

pipeline = Pipeline()
pipeline.add_component("producer", StringProducer())
pipeline.add_component("consumer_opt", StringConsumerOptional())

# fails:
pipeline.connect("producer.words", "consumer_opt.words")

[Pitrified]

Thank you for the fast response, that seems a far cleaner workaround than disabling validation

Just in case it's needed, I've narrowed it down to this line

    def run(self, words: List[str] | None = None) -> dict[str, Any]:  # noqa: UP006

where if I change from List to list it breaks.
Everywhere else I can use the native types.

Cheers!