LangChain tools in both directions¶

Tools bridge in both directions between common.ai’s toolsets and LangChain.

LangChain tools → ``AgentOperator``. No Airflow code is needed. pydantic-ai ships pydantic_ai.ext.langchain.LangChainToolset upstream, which wraps existing LangChain tools as an AbstractToolset. Drop it straight into AgentOperator:

from pydantic_ai.ext.langchain import LangChainToolset

AgentOperator(
    task_id="agent_with_langchain_tools",
    prompt="Research the question and summarise.",
    llm_conn_id="pydanticai_default",
    toolsets=[LangChainToolset([my_langchain_tool])],
)

common.ai toolsets → LangChain. The reverse direction is what airflow_toolset_to_langchain_tools() provides. It converts any pydantic-ai toolset – including SQLToolset, HookToolset, and MCPToolset – into a list of LangChain StructuredTool objects, so a LangChain agent or chain can call Airflow’s curated, connection-managed tools:

airflow/providers/common/ai/example_dags/example_langchain_toolset_bridge.py[source]

@dag(tags=["example"])
def example_langchain_toolset_bridge():
    """Run a LangChain SQL agent backed by Airflow's curated ``SQLToolset``."""

    @task
    def run_sql_agent(question: str = DEFAULT_QUESTION) -> str:
        from langchain.agents import create_agent

        from airflow.providers.common.ai.hooks.langchain import LangChainHook
        from airflow.providers.common.ai.toolsets import airflow_toolset_to_langchain_tools
        from airflow.providers.common.ai.toolsets.sql import SQLToolset

        # Airflow's curated, read-only SQL toolset, exposed as LangChain tools.
        # The bridge carries each tool's name, description, and args schema, and
        # routes calls back through SQLToolset (connection resolution + SQL
        # validation included).
        tools = airflow_toolset_to_langchain_tools(SQLToolset(db_conn_id=DB_CONN_ID))

        model = LangChainHook(llm_conn_id=LLM_CONN_ID, llm_model=LLM_MODEL).get_chat_model()
        agent = create_agent(
            model,
            tools=tools,
            system_prompt=(
                "You are a SQL analyst. Use list_tables and get_schema to explore "
                "the database, then run read-only queries to answer the question."
            ),
        )

        result = agent.invoke({"messages": [{"role": "user", "content": question}]})
        return result["messages"][-1].content

    run_sql_agent()

Each generated tool keeps the source tool’s name, description, and argument schema, and routes calls back through the original toolset, so the toolset’s own behavior (connection resolution, SQLToolset’s SQL validation, and allowed_tables filtering) still applies. get_tools runs eagerly at conversion time to enumerate the tools.

When a toolset raises pydantic-ai’s ModelRetry to ask the model to correct its input (SQLToolset does this on, for example, an unknown column), the bridge returns that message as the tool’s output so the model sees it and tries again. ModelRetry is a feed-the-model-and-retry signal rather than a failure, so returning it preserves the self-correction the toolset was written for and works no matter how the agent is configured to handle tool errors (raising would abort the run under create_agent’s default handling).

The bridge does not hold a toolset session open across calls: get_tools and every tool call each run under their own event loop, so for MCPToolset the connection is opened and torn down around each call. It reconnects per call, which is fine for stateless tools but unsuitable for stdio MCP servers (or any server that keeps state between calls), since each call starts a fresh session.

Note

Outside an agent run there is no live RunContext, so the bridge builds a minimal one with an inert placeholder model. The bundled toolsets ignore the context, so this is transparent for them. A custom toolset that reads live run state (ctx.model, ctx.messages, ctx.usage) will not behave correctly when bridged standalone.

Requires the langchain extra: pip install "apache-airflow-providers-common-ai[langchain]"