LangChain tools in both directions¶

Note

Experimental: this can change or be removed in a minor release of this provider. See Stable and experimental features.

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.

What a tool returns passes through Airflow’s secret masker before the model sees it.

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), or the model’s arguments fail the toolset’s validation, the model receives a LangChain ToolMessage with status="error" carrying the message, and can try again. The tool’s max_retries bounds this: once the model has failed that many times in a row, the call raises ToolCallError and the run ends. A refusal the toolset reports as final, by raising pydantic-ai’s ToolFailed, also reaches the model as an error result, without counting against that limit. Any other exception the toolset raises ends the run the same way.

The bridge does not hold a toolset session open across calls. On LangChain’s sync path every call runs on its own event loop, and even on the async path each call through MCPToolset opens and closes a session of its own. That 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.

A SandboxToolset bridged this way has to be used inside its with block, which provisions the sandbox on the first call and destroys it when the block ends.

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]"