priority

Raised when attempting to register a duplicate encryption/decryption handler.

Warning for beta features in LangGraph SDK.

Decide the priority settings of your LangGraph application's threads.

.. warning:: This API is in beta and may change in future versions.

To use, create a separate Python file and add the path to the file to your LangGraph API configuration file (langgraph.json). Within that file, create an instance of the Priority class and register an assign function with @priority.assign.

Example langgraph.json file:

{
  "dependencies": ["."],
  "graphs": {
    "agent": "./my_agent/agent.py:graph"
  },
  "env": ".env",
  "priority": {
    "path": "./priority.py:priority"
  }
}

The server calls the assign function at the first run on a thread with no settings, which can come long after the thread was created, and stores the settings it returns on the thread. Later runs on the thread use the stored settings. The function can be called more than once for one thread, for example after it raises, and only one result is stored, so keep it free of side effects.

from langgraph_sdk import Priority, PriorityContext, PrioritySettings

priority = Priority()

@priority.assign
async def assign(ctx: PriorityContext) -> PrioritySettings | None:
    return {
        "priority": 3 if "premium" in ctx.permissions else 0,
        # Stored unencrypted, so an opaque ID rather than personal data.
        "fairness_group": ctx.user["org_id"],
    }

Priority types for LangGraph.

This module defines the context the server passes to the function registered with @priority.assign, and the settings that function returns.