Improve run_api dynamic schema creation

[mpangrazzi]

This will add better description of run_api params / return type to dynamically create request / response pydantic models using docstring_parser to automatically parse run_api docstring.

For example, if you have a run_api like:

def run_api(question: str, urls: List[str]) -> str:
    """
    Ask a question about one or more websites using a Haystack pipeline.

    Args:
        question: The question to ask.
        urls: List of URLs to fetch.
        
    Returns:
        The answer to the question.
    """
    ...

It will add proper description, defaults (if present) and type to each param on pydantic model schema for question, urls and the return type.

This improved description will improve Hayhooks usage as OpenAPI Tool or MCP Tool.

[anakin87]

I would expect more failures with docstring parsing than with the previous simple approach.

What do you think?

• Do we want to force users to stick to this more rigid format?
• Should we instead provide a safe fallback if docstring parsing fails?

[mpangrazzi]

I think your suggestion makes sense, but honestly, I haven't been able to make docstring.parse() raise an exception in my tests 😅.

Even when mixing different docstring formats (or not strictly adhering to them), it doesn't raise any exceptions. Instead, it just skips certain parameters. If, for example, you type ((int), you might get a slightly malformed type_name value, but the function doesn't fail -it simply behaves unexpectedly. Evaluating this behaviour automatically could be tricky.

I also noticed that within Haystack, .parse() exceptions are not explicitly handled either (in ComponentTool).

Perhaps we can leave the code as it is for now unless a bug pops out in the future?

[mpangrazzi]

I am also attaching an example: if you provide a run_api like the following:

def run_api(self, urls: List[str], question: str) -> str:
        """
        Ask a question to a list of websites.

        Args:
            urls (List[str]): List of URLs to ask the question to.
            question (str): Question to ask the websites.

        Returns:
            str: Answer to the question.
        """
        result = self.pipeline.run({"fetcher": {"urls": urls}, "prompt": {"query": question}})
        return result["llm"]["replies"][0]

You'll get a Tool description like the following (using MCP Inspector):