test_filter_injected_args_async() — langchain Function Reference

Architecture documentation for the test_filter_injected_args_async() function in test_tools.py from the langchain codebase.

Function python LangChainCore MessageInterface

Entity Profile

LangChainCore→ MessageInterface→ test_filter_injected_args_async() — langchain Function Reference

Dependency Diagram

graph TD
  93b50798_ae0e_b762_7d7d_e0ad6260e3f4["test_filter_injected_args_async()"]
  8e7836ae_e72c_f670_72a5_4ca6d46e3555["test_tools.py"]
  93b50798_ae0e_b762_7d7d_e0ad6260e3f4 -->|defined in| 8e7836ae_e72c_f670_72a5_4ca6d46e3555
  style 93b50798_ae0e_b762_7d7d_e0ad6260e3f4 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/test_tools.py lines 3130–3161

async def test_filter_injected_args_async() -> None:
    """Test that injected args are filtered in async tool execution."""

    @tool
    async def async_search_tool(
        query: str,
        state: Annotated[dict, InjectedToolArg()],
    ) -> str:
        """Async search with injected state.

        Args:
            query: The search query.
            state: Injected state context.
        """
        return f"Async results for: {query}"

    handler = CallbackHandlerWithInputCapture(captured_inputs=[])
    result = await async_search_tool.ainvoke(
        {"query": "async test", "state": {"user_id": 456}},
        config={"callbacks": [handler]},
    )

    assert result == "Async results for: async test"
    assert handler.tool_starts == 1
    assert len(handler.captured_inputs) == 1

    # Verify filtering in async execution
    captured = handler.captured_inputs[0]
    assert captured is not None
    assert "query" in captured
    assert "state" not in captured
    assert captured["query"] == "async test"