Add internal shared core for the LangChain-family plugins

pyproject.toml

@@ -32,6 +32,11 @@ openai-agents = ["openai-agents>=0.17.5", "mcp>=1.9.4, <2"]
 google-adk = ["google-adk>=2.2.0,<3"]
 langgraph = ["langgraph>=1.1.0"]
 langsmith = ["langsmith>=0.7.34,<0.9"]
+deepagents = [
+  "deepagents>=0.6.12,<0.7; python_version >= '3.11'",
+  "langchain>=1.3.11,<2; python_version >= '3.11'",
+  "langchain-core>=1.4.8,<2; python_version >= '3.11'",
+]
 lambda-worker-otel = [
   "opentelemetry-api>=1.11.1,<2",
   "opentelemetry-sdk>=1.11.1,<2",
@@ -81,6 +86,10 @@ dev = [
   "moto[s3,server]>=5",
   "langgraph>=1.1.0",
   "langsmith>=0.7.34,<0.9",
+  "deepagents>=0.6.12,<0.7; python_version >= '3.11'",
+  "langchain>=1.3.11,<2; python_version >= '3.11'",
+  "langchain-core>=1.4.8,<2; python_version >= '3.11'",
+  "langchain-anthropic>=1.4.7; python_version >= '3.11'",
   "setuptools<82",
   "opentelemetry-exporter-otlp-proto-grpc>=1.11.1,<2",
   "opentelemetry-semantic-conventions>=0.40b0,<1",

temporalio/contrib/_langchain/__init__.py

@@ -0,0 +1,10 @@
+"""Internal shared machinery for the LangChain-family plugins.
+
+This package is shared infrastructure for ``temporalio.contrib.langgraph``,
+``temporalio.contrib.langsmith``, and ``temporalio.contrib.deepagents``. It is
+NOT a public API: names, modules, and behavior may change without notice.
+
+Import discipline: this ``__init__`` performs no imports, and submodules defer
+every third-party import into function bodies, so the package is importable
+with none of the LangChain-family distributions installed.
+"""

temporalio/contrib/_langchain/_activity_helpers.py

@@ -0,0 +1,90 @@
+"""Activity-side helpers shared by the LangChain-family plugins."""
+
+from __future__ import annotations
+
+import asyncio
+from datetime import timedelta
+from functools import wraps
+from typing import Any, Callable
+
+from temporalio import activity
+from temporalio.exceptions import ApplicationError
+
+
+def auto_heartbeater(fn: Callable) -> Callable:
+    """Heartbeat at half the configured ``heartbeat_timeout`` while ``fn`` runs.
+
+    Long LLM calls (thinking mode, long context, streaming accumulation) can run
+    well past a scheduler's patience; without a heartbeat Temporal would cancel
+    them and surface a ``HeartbeatTimeoutError`` instead of the real problem.
+    """
+
+    @wraps(fn)
+    async def wrapped(*args: Any, **kwargs: Any) -> Any:
+        heartbeat_timeout = activity.info().heartbeat_timeout
+        beat_task: asyncio.Task | None = None
+        if heartbeat_timeout:
+            interval = heartbeat_timeout.total_seconds() / 2
+
+            async def beat() -> None:
+                while True:
+                    activity.heartbeat()
+                    await asyncio.sleep(interval)
+
+            beat_task = asyncio.create_task(beat())
+        try:
+            return await fn(*args, **kwargs)
+        finally:
+            if beat_task is not None:
+                beat_task.cancel()
+                # Let the cancellation land before returning so no pending task
+                # outlives the activity (a bare ``cancel()`` leaves the task to
+                # be destroyed while pending if the loop shuts down first).
+                # ``asyncio.wait`` never re-raises the task's CancelledError.
+                await asyncio.wait([beat_task])
+
+    return wrapped
+
+
+def translate_api_error(exc: Exception) -> ApplicationError | None:
+    """Map an LLM SDK HTTP error onto Temporal's retry contract.
+
+    Works by duck typing so neither ``openai`` nor ``anthropic`` needs to be
+    imported here: both expose ``status_code`` and ``response.headers``. Returns
+    ``None`` when ``exc`` is not a recognizable HTTP status error, so the caller
+    can fall through to its generic handling.
+    """
+    status = getattr(exc, "status_code", None)
+    if status is None:
+        return None
+    headers: dict[str, Any] = {}
+    response = getattr(exc, "response", None)
+    if response is not None:
+        headers = dict(getattr(response, "headers", {}) or {})
+    # Case-insensitive header access.
+    lower = {str(k).lower(): v for k, v in headers.items()}
+
+    retryable = status in (408, 409, 429) or 500 <= status < 600
+    should_retry = lower.get("x-should-retry")
+    if should_retry == "false":
+        retryable = False
+    elif should_retry == "true":
+        retryable = True
+
+    delay_ms = lower.get("retry-after-ms")
+    retry_after = lower.get("retry-after")
+    next_delay: timedelta | None = None
+    try:
+        if delay_ms is not None:
+            next_delay = timedelta(milliseconds=int(delay_ms))
+        elif retry_after is not None:
+            next_delay = timedelta(seconds=int(retry_after))
+    except (TypeError, ValueError):
+        next_delay = None
+
+    return ApplicationError(
+        str(exc),
+        type=type(exc).__name__,
+        non_retryable=not retryable,
+        next_retry_delay=next_delay,
+    )

temporalio/contrib/_langchain/_aio_to_thread.py

@@ -0,0 +1,59 @@
+"""LangSmith ``aio_to_thread`` override shared by the LangChain-family plugins."""
+
+from __future__ import annotations
+
+from typing import Any, Callable
+
+import temporalio.workflow
+
+_installed = False
+
+
+async def _temporal_aio_to_thread(
+    default_aio_to_thread: Callable[..., Any],
+    ctx: Any,
+    func: Callable[..., Any],
+    /,
+    *args: Any,
+    **kwargs: Any,
+) -> Any:
+    """Run LangSmith's ``aio_to_thread`` synchronously inside Temporal workflows.
+
+    The ``@traceable`` decorator on async functions uses ``aio_to_thread()`` →
+    ``loop.run_in_executor()`` for run setup/teardown. The Temporal workflow
+    event loop does not support ``run_in_executor``. This override runs those
+    functions synchronously on the workflow thread when inside a workflow,
+    and delegates to the default implementation outside workflows.
+
+    Registered via ``langsmith.set_runtime_overrides(aio_to_thread=...)``.
+    """
+    if not temporalio.workflow.in_workflow():
+        return await default_aio_to_thread(ctx, func, *args, **kwargs)
+    with temporalio.workflow.unsafe.sandbox_unrestricted():
+        return ctx.run(func, *args, **kwargs)
+
+
+def install_aio_to_thread_override() -> None:
+    """Install the ``aio_to_thread`` override via LangSmith's official API.
+
+    Safe to call multiple times and from multiple plugins; the override is
+    installed once per process. It is deliberately never uninstalled:
+    LangSmith exposes a single process-wide override slot (each
+    ``set_runtime_overrides`` call replaces it wholesale), so resetting it on
+    one worker's shutdown would strip a composed plugin's still-needed
+    override. Leaving it installed is safe — the override defers to
+    LangSmith's default thread hop whenever ``workflow.in_workflow()`` is
+    false, so it is inert outside workflows.
+
+    Raises whatever the lazy ``langsmith`` import or
+    ``set_runtime_overrides`` call raises (e.g. ``ImportError`` when
+    LangSmith is absent); the installed flag stays unset on failure so a
+    later call can retry.
+    """
+    global _installed  # noqa: PLW0603
+    if _installed:
+        return
+    import langsmith
+
+    langsmith.set_runtime_overrides(aio_to_thread=_temporal_aio_to_thread)
+    _installed = True

temporalio/contrib/_langchain/_converter.py

@@ -0,0 +1,61 @@
+"""Payload converter shared by the LangChain-family plugins.
+
+This module eagerly imports ``temporalio.contrib.pydantic``; only plugins
+that actually wire the converter import it, so the core package itself stays
+importable without pydantic installed.
+"""
+
+from __future__ import annotations
+
+import dataclasses
+
+from temporalio.contrib.pydantic import PydanticPayloadConverter, ToJsonOptions
+from temporalio.converter import DataConverter
+
+
+class LangChainPayloadConverter(PydanticPayloadConverter):
+    """Pydantic payload converter pinned to ``exclude_unset=True``.
+
+    LangChain request/response types are deeply nested with many
+    ``Optional[...] = None`` fields. Shipping every unset default inflates
+    payloads several-fold and some peers reject the explicit nulls on
+    round-trip, so unset fields are excluded by convention.
+    """
+
+    def __init__(self) -> None:
+        """Construct the converter with ``exclude_unset`` serialization."""
+        super().__init__(ToJsonOptions(exclude_unset=True))
+
+
+data_converter = DataConverter(payload_converter_class=LangChainPayloadConverter)
+"""The family default data converter (LangChain messages are shipped as their
+``dumpd`` JSON form, so the Pydantic converter only ever sees plain
+containers)."""
+
+
+def build_data_converter(
+    user_converter: DataConverter | None,
+    *,
+    plugin_name: str,
+) -> DataConverter:
+    """Compose the family converter with whatever the caller already set.
+
+    * ``None`` — install the family default.
+    * the SDK default converter — swap in the LangChain-aware Pydantic
+      converter via :func:`dataclasses.replace`.
+    * a custom converter — refuse rather than silently clobber it; the caller
+      must fold ``LangChainPayloadConverter`` into their own converter.
+      ``plugin_name`` names the refusing plugin in the error.
+    """
+    if user_converter is None:
+        return data_converter
+    if user_converter is DataConverter.default:
+        return dataclasses.replace(
+            user_converter, payload_converter_class=LangChainPayloadConverter
+        )
+    raise ValueError(
+        f"{plugin_name} cannot compose with a custom data_converter "
+        "automatically. Set payload_converter_class=LangChainPayloadConverter "
+        "on your own DataConverter (so LangChain messages serialize with "
+        "exclude_unset=True), or omit data_converter to use the plugin default."
+    )

temporalio/contrib/_langchain/_messages.py

@@ -0,0 +1,44 @@
+"""LangChain object serialization shared by the LangChain-family plugins."""
+
+from __future__ import annotations
+
+from typing import Any
+
+
+def dump_object(obj: Any) -> Any:
+    """Serialize a single LangChain ``Serializable`` (message, tool call, …)."""
+    from langchain_core.load import dumpd
+
+    return dumpd(obj)
+
+
+def load_object(data: Any) -> Any:
+    """Rehydrate a value produced by :func:`dump_object`, preserving subtype."""
+    from langchain_core.load import load
+
+    return load(data)
+
+
+def dump_messages(messages: Any) -> list[Any]:
+    """Serialize a sequence of LangChain messages to their ``dumpd`` form."""
+    from langchain_core.load import dumpd
+
+    return [dumpd(m) for m in messages]
+
+
+def load_messages(dumped: list[Any]) -> list[Any]:
+    """Rehydrate messages serialized by :func:`dump_messages`."""
+    from langchain_core.load import load
+
+    return [load(d) for d in dumped]
+
+
+def tool_to_schema(tool: Any) -> dict[str, Any]:
+    """Advertise a tool to the model as a full OpenAI tool schema.
+
+    Carries name + description + argument JSON schema so the model can build
+    valid arguments, not just select the tool by name.
+    """
+    from langchain_core.utils.function_calling import convert_to_openai_tool
+
+    return convert_to_openai_tool(tool)

temporalio/contrib/_langchain/_passthrough.py

@@ -0,0 +1,22 @@
+"""Sandbox passthrough-list merging shared by the LangChain-family plugins.
+
+Only the MECHANISM lives here; each plugin owns its module list (the lists
+are behavior for released plugins and must not drift by sharing).
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterable, Sequence
+
+
+def merge_passthrough_modules(
+    defaults: Sequence[str], user: Iterable[str] | None
+) -> tuple[str, ...]:
+    """Merge caller-supplied passthrough modules with a plugin's defaults.
+
+    Order-preserving: defaults first, then user additions, first occurrence
+    wins on duplicates.
+    """
+    merged = [*defaults, *(user or ())]
+    # dict.fromkeys preserves order while de-duplicating.
+    return tuple(dict.fromkeys(merged))