Home - News ( United States | United Kingdom | Italy | Germany ) - Football scores

Showing content from https://langchain-ai.github.io/langgraph/cloud/reference/sdk/python_sdk_ref/ below:

SDK (Python)

The LangGraph client implementations connect to the LangGraph API.

This module provides both asynchronous (get_client(url="http://localhost:2024")) or LangGraphClient) and synchronous (get_sync_client(url="http://localhost:2024")) or SyncLanggraphClient) clients to interacting with the LangGraph API's core resources such as Assistants, Threads, Runs, and Cron jobs, as well as its persistent document Store.

Classes:

Functions:

Top-level client for LangGraph API.

Attributes:

Manages versioned configuration for your graphs.

Handles (potentially) multi-turn interactions, such as conversational threads.

Controls individual invocations of the graph.

Manages scheduled operations.

Interfaces with persistent, shared data storage.

Handle async requests to the LangGraph API.

Adds additional error messaging & content handling above the provided httpx client.

Attributes:

Underlying HTTPX async client.

Methods:

Send a GET request.

Send a POST request.

Send a PUT request.

Send a PATCH request.

Send a DELETE request.

Stream results using SSE.

get(
    path: str,
    *,
    params: QueryParamTypes | None = None,
    headers: dict[str, str] | None = None,
    on_response: Callable[[Response], None] | None = None
) -> Any

Send a GET request.

post(
    path: str,
    *,
    json: dict | None,
    headers: dict[str, str] | None = None,
    on_response: Callable[[Response], None] | None = None
) -> Any

Send a POST request.

Send a PUT request.

Send a PATCH request.

delete(
    path: str,
    *,
    json: Any | None = None,
    headers: dict[str, str] | None = None,
    on_response: Callable[[Response], None] | None = None
) -> None

Send a DELETE request.

Stream results using SSE.

Client for managing assistants in LangGraph.

This class provides methods to interact with assistants, which are versioned configurations of your graph.

client = get_client(url="http://localhost:2024")
assistant = await client.assistants.get("assistant_id_123")

Methods:

Get an assistant by ID.

Parameters:

The ID of the assistant to get.

Optional custom headers to include with the request.

Returns:

Assistant Object.

assistant = await client.assistants.get(
    assistant_id="my_assistant_id"
)
print(assistant)

----------------------------------------------------

{
    'assistant_id': 'my_assistant_id',
    'graph_id': 'agent',
    'created_at': '2024-06-25T17:10:33.109781+00:00',
    'updated_at': '2024-06-25T17:10:33.109781+00:00',
    'config': {},
    'metadata': {'created_by': 'system'},
    'version': 1,
    'name': 'my_assistant'
}

Get the graph of an assistant by ID.

Parameters:

The ID of the assistant to get the graph of.

Include graph representation of subgraphs. If an integer value is provided, only subgraphs with a depth less than or equal to the value will be included.

Optional custom headers to include with the request.

Returns:

The graph information for the assistant in JSON format.

client = get_client(url="http://localhost:2024")
graph_info = await client.assistants.get_graph(
    assistant_id="my_assistant_id"
)
print(graph_info)

--------------------------------------------------------------------------------------------------------------------------

{
    'nodes':
        [
            {'id': '__start__', 'type': 'schema', 'data': '__start__'},
            {'id': '__end__', 'type': 'schema', 'data': '__end__'},
            {'id': 'agent','type': 'runnable','data': {'id': ['langgraph', 'utils', 'RunnableCallable'],'name': 'agent'}},
        ],
    'edges':
        [
            {'source': '__start__', 'target': 'agent'},
            {'source': 'agent','target': '__end__'}
        ]
}

Get the schemas of an assistant by ID.

Parameters:

The ID of the assistant to get the schema of.

Optional custom headers to include with the request.

Returns:

The graph schema for the assistant.

client = get_client(url="http://localhost:2024")
schema = await client.assistants.get_schemas(
    assistant_id="my_assistant_id"
)
print(schema)

----------------------------------------------------------------------------------------------------------------------------

{
    'graph_id': 'agent',
    'state_schema':
        {
            'title': 'LangGraphInput',
            '$ref': '#/definitions/AgentState',
            'definitions':
                {
                    'BaseMessage':
                        {
                            'title': 'BaseMessage',
                            'description': 'Base abstract Message class. Messages are the inputs and outputs of ChatModels.',
                            'type': 'object',
                            'properties':
                                {
                                 'content':
                                    {
                                        'title': 'Content',
                                        'anyOf': [
                                            {'type': 'string'},
                                            {'type': 'array','items': {'anyOf': [{'type': 'string'}, {'type': 'object'}]}}
                                        ]
                                    },
                                'additional_kwargs':
                                    {
                                        'title': 'Additional Kwargs',
                                        'type': 'object'
                                    },
                                'response_metadata':
                                    {
                                        'title': 'Response Metadata',
                                        'type': 'object'
                                    },
                                'type':
                                    {
                                        'title': 'Type',
                                        'type': 'string'
                                    },
                                'name':
                                    {
                                        'title': 'Name',
                                        'type': 'string'
                                    },
                                'id':
                                    {
                                        'title': 'Id',
                                        'type': 'string'
                                    }
                                },
                            'required': ['content', 'type']
                        },
                    'AgentState':
                        {
                            'title': 'AgentState',
                            'type': 'object',
                            'properties':
                                {
                                    'messages':
                                        {
                                            'title': 'Messages',
                                            'type': 'array',
                                            'items': {'$ref': '#/definitions/BaseMessage'}
                                        }
                                },
                            'required': ['messages']
                        }
                }
        },
    'context_schema':
        {
            'title': 'Context',
            'type': 'object',
            'properties':
                {
                    'model_name':
                        {
                            'title': 'Model Name',
                            'enum': ['anthropic', 'openai'],
                            'type': 'string'
                        }
                }
        }
}

get_subgraphs(
    assistant_id: str,
    namespace: str | None = None,
    recurse: bool = False,
    *,
    headers: dict[str, str] | None = None
) -> Subgraphs

Get the schemas of an assistant by ID.

Parameters:

The ID of the assistant to get the schema of.

Optional namespace to filter by.

Whether to recursively get subgraphs.

Optional custom headers to include with the request.

Returns:

The graph schema for the assistant.

create(
    graph_id: str | None,
    config: Config | None = None,
    *,
    context: Context | None = None,
    metadata: Json = None,
    assistant_id: str | None = None,
    if_exists: OnConflictBehavior | None = None,
    name: str | None = None,
    headers: dict[str, str] | None = None,
    description: str | None = None
) -> Assistant

Create a new assistant.

Useful when graph is configurable and you want to create different assistants based on different configurations.

Parameters:

The ID of the graph the assistant should use. The graph ID is normally set in your langgraph.json configuration.

Configuration to use for the graph.

Metadata to add to assistant.

Static context to add to the assistant.

Supported with langgraph>=0.6.0

Assistant ID to use, will default to a random UUID if not provided.

How to handle duplicate creation. Defaults to 'raise' under the hood. Must be either 'raise' (raise error if duplicate), or 'do_nothing' (return existing assistant).

The name of the assistant. Defaults to 'Untitled' under the hood.

Optional custom headers to include with the request.

Optional description of the assistant. The description field is available for langgraph-api server version>=0.0.45

Returns:

The created assistant.

client = get_client(url="http://localhost:2024")
assistant = await client.assistants.create(
    graph_id="agent",
    context={"model_name": "openai"},
    metadata={"number":1},
    assistant_id="my-assistant-id",
    if_exists="do_nothing",
    name="my_name"
)

update(
    assistant_id: str,
    *,
    graph_id: str | None = None,
    config: Config | None = None,
    context: Context | None = None,
    metadata: Json = None,
    name: str | None = None,
    headers: dict[str, str] | None = None,
    description: str | None = None
) -> Assistant

Update an assistant.

Use this to point to a different graph, update the configuration, or change the metadata of an assistant.

Parameters:

Assistant to update.

The ID of the graph the assistant should use. The graph ID is normally set in your langgraph.json configuration. If None, assistant will keep pointing to same graph.

Configuration to use for the graph.

Static context to add to the assistant.

Supported with langgraph>=0.6.0

Metadata to merge with existing assistant metadata.

The new name for the assistant.

Optional custom headers to include with the request.

Optional description of the assistant. The description field is available for langgraph-api server version>=0.0.45

Returns:

The updated assistant.

client = get_client(url="http://localhost:2024")
assistant = await client.assistants.update(
    assistant_id='e280dad7-8618-443f-87f1-8e41841c180f',
    graph_id="other-graph",
    context={"model_name": "anthropic"},
    metadata={"number":2}
)

delete(
    assistant_id: str,
    *,
    headers: dict[str, str] | None = None
) -> None

Delete an assistant.

Parameters:

The assistant ID to delete.

Optional custom headers to include with the request.

Returns:

None

client = get_client(url="http://localhost:2024")
await client.assistants.delete(
    assistant_id="my_assistant_id"
)

Search for assistants.

Parameters:

Metadata to filter by. Exact match filter for each KV pair.

The ID of the graph to filter by. The graph ID is normally set in your langgraph.json configuration.

The maximum number of results to return.

The number of results to skip.

The field to sort by.

The order to sort by.

Optional custom headers to include with the request.

Returns:

list[Assistant]: A list of assistants.

client = get_client(url="http://localhost:2024")
assistants = await client.assistants.search(
    metadata = {"name":"my_name"},
    graph_id="my_graph_id",
    limit=5,
    offset=5
)

List all versions of an assistant.

Parameters:

The assistant ID to get versions for.

Metadata to filter versions by. Exact match filter for each KV pair.

The maximum number of versions to return.

The number of versions to skip.

Optional custom headers to include with the request.

Returns:

client = get_client(url="http://localhost:2024")
assistant_versions = await client.assistants.get_versions(
    assistant_id="my_assistant_id"
)

Change the version of an assistant.

Parameters:

The assistant ID to delete.

The version to change to.

Optional custom headers to include with the request.

Returns:

Assistant Object.

client = get_client(url="http://localhost:2024")
new_version_assistant = await client.assistants.set_latest(
    assistant_id="my_assistant_id",
    version=3
)

Client for managing threads in LangGraph.

A thread maintains the state of a graph across multiple interactions/invocations (aka runs). It accumulates and persists the graph's state, allowing for continuity between separate invocations of the graph.

client = get_client(url="http://localhost:2024"))
new_thread = await client.threads.create(metadata={"user_id": "123"})

Methods:

Get a thread by ID.

Parameters:

The ID of the thread to get.

Optional custom headers to include with the request.

Returns:

Thread object.

client = get_client(url="http://localhost:2024")
thread = await client.threads.get(
    thread_id="my_thread_id"
)
print(thread)

-----------------------------------------------------

{
    'thread_id': 'my_thread_id',
    'created_at': '2024-07-18T18:35:15.540834+00:00',
    'updated_at': '2024-07-18T18:35:15.540834+00:00',
    'metadata': {'graph_id': 'agent'}
}

Create a new thread.

Parameters:

Metadata to add to thread.

ID of thread. If None, ID will be a randomly generated UUID.

How to handle duplicate creation. Defaults to 'raise' under the hood. Must be either 'raise' (raise error if duplicate), or 'do_nothing' (return existing thread).

Apply a list of supersteps when creating a thread, each containing a sequence of updates. Each update has values or command and as_node. Used for copying a thread between deployments.

Optional graph ID to associate with the thread.

Optional custom headers to include with the request.

Returns:

The created thread.

client = get_client(url="http://localhost:2024")
thread = await client.threads.create(
    metadata={"number":1},
    thread_id="my-thread-id",
    if_exists="raise"
)

Update a thread.

Parameters:

ID of thread to update.

Metadata to merge with existing thread metadata.

Optional custom headers to include with the request.

Returns:

The created thread.

client = get_client(url="http://localhost:2024")
thread = await client.threads.update(
    thread_id="my-thread-id",
    metadata={"number":1},
)

delete(
    thread_id: str, *, headers: dict[str, str] | None = None
) -> None

Delete a thread.

Parameters:

The ID of the thread to delete.

Optional custom headers to include with the request.

Returns:

None

client = get_client(url="http://localhost2024)
await client.threads.delete(
    thread_id="my_thread_id"
)

Search for threads.

Parameters:

Thread metadata to filter on.

State values to filter on.

Thread status to filter on. Must be one of 'idle', 'busy', 'interrupted' or 'error'.

Limit on number of threads to return.

Offset in threads table to start search from.

Sort by field.

Sort order.

Optional custom headers to include with the request.

Returns:

list[Thread]: List of the threads matching the search parameters.

client = get_client(url="http://localhost:2024")
threads = await client.threads.search(
    metadata={"number":1},
    status="interrupted",
    limit=15,
    offset=5
)

copy(
    thread_id: str, *, headers: dict[str, str] | None = None
) -> None

Copy a thread.

Parameters:

The ID of the thread to copy.

Optional custom headers to include with the request.

Returns:

None

client = get_client(url="http://localhost:2024)
await client.threads.copy(
    thread_id="my_thread_id"
)

Get the state of a thread.

Parameters:

The ID of the thread to get the state of.

The checkpoint to get the state of.

(deprecated) The checkpoint ID to get the state of.

Include subgraphs states.

Optional custom headers to include with the request.

Returns:

the thread of the state.

client = get_client(url="http://localhost:2024)
thread_state = await client.threads.get_state(
    thread_id="my_thread_id",
    checkpoint_id="my_checkpoint_id"
)
print(thread_state)

----------------------------------------------------------------------------------------------------------------------------------------------------------------------

{
    'values': {
        'messages': [
            {
                'content': 'how are you?',
                'additional_kwargs': {},
                'response_metadata': {},
                'type': 'human',
                'name': None,
                'id': 'fe0a5778-cfe9-42ee-b807-0adaa1873c10',
                'example': False
            },
            {
                'content': "I'm doing well, thanks for asking! I'm an AI assistant created by Anthropic to be helpful, honest, and harmless.",
                'additional_kwargs': {},
                'response_metadata': {},
                'type': 'ai',
                'name': None,
                'id': 'run-159b782c-b679-4830-83c6-cef87798fe8b',
                'example': False,
                'tool_calls': [],
                'invalid_tool_calls': [],
                'usage_metadata': None
            }
        ]
    },
    'next': [],
    'checkpoint':
        {
            'thread_id': 'e2496803-ecd5-4e0c-a779-3226296181c2',
            'checkpoint_ns': '',
            'checkpoint_id': '1ef4a9b8-e6fb-67b1-8001-abd5184439d1'
        }
    'metadata':
        {
            'step': 1,
            'run_id': '1ef4a9b8-d7da-679a-a45a-872054341df2',
            'source': 'loop',
            'writes':
                {
                    'agent':
                        {
                            'messages': [
                                {
                                    'id': 'run-159b782c-b679-4830-83c6-cef87798fe8b',
                                    'name': None,
                                    'type': 'ai',
                                    'content': "I'm doing well, thanks for asking! I'm an AI assistant created by Anthropic to be helpful, honest, and harmless.",
                                    'example': False,
                                    'tool_calls': [],
                                    'usage_metadata': None,
                                    'additional_kwargs': {},
                                    'response_metadata': {},
                                    'invalid_tool_calls': []
                                }
                            ]
                        }
                },
    'user_id': None,
    'graph_id': 'agent',
    'thread_id': 'e2496803-ecd5-4e0c-a779-3226296181c2',
    'created_by': 'system',
    'assistant_id': 'fe096781-5601-53d2-b2f6-0d3403f7e9ca'},
    'created_at': '2024-07-25T15:35:44.184703+00:00',
    'parent_config':
        {
            'thread_id': 'e2496803-ecd5-4e0c-a779-3226296181c2',
            'checkpoint_ns': '',
            'checkpoint_id': '1ef4a9b8-d80d-6fa7-8000-9300467fad0f'
        }
}

Update the state of a thread.

Parameters:

The ID of the thread to update.

The values to update the state with.

Update the state as if this node had just executed.

The checkpoint to update the state of.

(deprecated) The checkpoint ID to update the state of.

Optional custom headers to include with the request.

Returns:

client = get_client(url="http://localhost:2024)
response = await client.threads.update_state(
    thread_id="my_thread_id",
    values={"messages":[{"role": "user", "content": "hello!"}]},
    as_node="my_node",
)
print(response)

----------------------------------------------------------------------------------------------------------------------------------------------------------------------

{
    'checkpoint': {
        'thread_id': 'e2496803-ecd5-4e0c-a779-3226296181c2',
        'checkpoint_ns': '',
        'checkpoint_id': '1ef4a9b8-e6fb-67b1-8001-abd5184439d1',
        'checkpoint_map': {}
    }
}

Get the state history of a thread.

Parameters:

The ID of the thread to get the state history for.

Return states for this subgraph. If empty defaults to root.

The maximum number of states to return.

Return states before this checkpoint.

Filter states by metadata key-value pairs.

Optional custom headers to include with the request.

Returns:

list[ThreadState]: the state history of the thread.

client = get_client(url="http://localhost:2024)
thread_state = await client.threads.get_history(
    thread_id="my_thread_id",
    limit=5,
)

Client for managing runs in LangGraph.

A run is a single assistant invocation with optional input, config, context, and metadata. This client manages runs, which can be stateful (on threads) or stateless.

client = get_client(url="http://localhost:2024")
run = await client.runs.create(assistant_id="asst_123", thread_id="thread_456", input={"query": "Hello"})

Methods:

Create a run and stream the results.

Create a background run.

Create a batch of stateless background runs.

Create a run, wait until it finishes and return the final state.

List runs.

Get a run.

Get a run.

Block until a run is done. Returns the final state of the thread.

Stream output from a run in real-time, until the run is done.

Delete a run.

stream(
    thread_id: str | None,
    assistant_id: str,
    *,
    input: dict | None = None,
    command: Command | None = None,
    stream_mode: (
        StreamMode | Sequence[StreamMode]
    ) = "values",
    stream_subgraphs: bool = False,
    stream_resumable: bool = False,
    metadata: dict | None = None,
    config: Config | None = None,
    context: Context | None = None,
    checkpoint: Checkpoint | None = None,
    checkpoint_id: str | None = None,
    checkpoint_during: bool | None = None,
    interrupt_before: All | Sequence[str] | None = None,
    interrupt_after: All | Sequence[str] | None = None,
    feedback_keys: Sequence[str] | None = None,
    on_disconnect: DisconnectMode | None = None,
    on_completion: OnCompletionBehavior | None = None,
    webhook: str | None = None,
    multitask_strategy: MultitaskStrategy | None = None,
    if_not_exists: IfNotExists | None = None,
    after_seconds: int | None = None,
    headers: dict[str, str] | None = None,
    on_run_created: (
        Callable[[RunCreateMetadata], None] | None
    ) = None
) -> AsyncIterator[StreamPart]

Create a run and stream the results.

Parameters:

the thread ID to assign to the thread. If None will create a stateless run.

The assistant ID or graph name to stream from. If using graph name, will default to first assistant created from that graph.

The input to the graph.

A command to execute. Cannot be combined with input.

The stream mode(s) to use.

Whether to stream output from subgraphs.

Whether the stream is considered resumable. If true, the stream can be resumed and replayed in its entirety even after disconnection.

Metadata to assign to the run.

The configuration for the assistant.

Static context to add to the assistant.

Supported with langgraph>=0.6.0

The checkpoint to resume from.

Whether to checkpoint during the run (or only at the end/interruption).

Nodes to interrupt immediately before they get executed.

Nodes to Nodes to interrupt immediately after they get executed.

Feedback keys to assign to run.

The disconnect mode to use. Must be one of 'cancel' or 'continue'.

Whether to delete or keep the thread created for a stateless run. Must be one of 'delete' or 'keep'.

Webhook to call after LangGraph API call is done.

Multitask strategy to use. Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'.

How to handle missing thread. Defaults to 'reject'. Must be either 'reject' (raise error if missing), or 'create' (create new thread).

The number of seconds to wait before starting the run. Use to schedule future runs.

Callback when a run is created.

Returns:

client = get_client(url="http://localhost:2024)
async for chunk in client.runs.stream(
    thread_id=None,
    assistant_id="agent",
    input={"messages": [{"role": "user", "content": "how are you?"}]},
    stream_mode=["values","debug"],
    metadata={"name":"my_run"},
    context={"model_name": "anthropic"},
    interrupt_before=["node_to_stop_before_1","node_to_stop_before_2"],
    interrupt_after=["node_to_stop_after_1","node_to_stop_after_2"],
    feedback_keys=["my_feedback_key_1","my_feedback_key_2"],
    webhook="https://my.fake.webhook.com",
    multitask_strategy="interrupt"
):
    print(chunk)

------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------

StreamPart(event='metadata', data={'run_id': '1ef4a9b8-d7da-679a-a45a-872054341df2'})
StreamPart(event='values', data={'messages': [{'content': 'how are you?', 'additional_kwargs': {}, 'response_metadata': {}, 'type': 'human', 'name': None, 'id': 'fe0a5778-cfe9-42ee-b807-0adaa1873c10', 'example': False}]})
StreamPart(event='values', data={'messages': [{'content': 'how are you?', 'additional_kwargs': {}, 'response_metadata': {}, 'type': 'human', 'name': None, 'id': 'fe0a5778-cfe9-42ee-b807-0adaa1873c10', 'example': False}, {'content': "I'm doing well, thanks for asking! I'm an AI assistant created by Anthropic to be helpful, honest, and harmless.", 'additional_kwargs': {}, 'response_metadata': {}, 'type': 'ai', 'name': None, 'id': 'run-159b782c-b679-4830-83c6-cef87798fe8b', 'example': False, 'tool_calls': [], 'invalid_tool_calls': [], 'usage_metadata': None}]})
StreamPart(event='end', data=None)

create(
    thread_id: str | None,
    assistant_id: str,
    *,
    input: dict | None = None,
    command: Command | None = None,
    stream_mode: (
        StreamMode | Sequence[StreamMode]
    ) = "values",
    stream_subgraphs: bool = False,
    stream_resumable: bool = False,
    metadata: dict | None = None,
    config: Config | None = None,
    context: Context | None = None,
    checkpoint: Checkpoint | None = None,
    checkpoint_id: str | None = None,
    checkpoint_during: bool | None = None,
    interrupt_before: All | Sequence[str] | None = None,
    interrupt_after: All | Sequence[str] | None = None,
    webhook: str | None = None,
    multitask_strategy: MultitaskStrategy | None = None,
    if_not_exists: IfNotExists | None = None,
    on_completion: OnCompletionBehavior | None = None,
    after_seconds: int | None = None,
    headers: dict[str, str] | None = None,
    on_run_created: (
        Callable[[RunCreateMetadata], None] | None
    ) = None
) -> Run

Create a background run.

Parameters:

the thread ID to assign to the thread. If None will create a stateless run.

The assistant ID or graph name to stream from. If using graph name, will default to first assistant created from that graph.

The input to the graph.

A command to execute. Cannot be combined with input.

The stream mode(s) to use.

Whether to stream output from subgraphs.

Whether the stream is considered resumable. If true, the stream can be resumed and replayed in its entirety even after disconnection.

Metadata to assign to the run.

The configuration for the assistant.

Static context to add to the assistant.

Supported with langgraph>=0.6.0

The checkpoint to resume from.

Whether to checkpoint during the run (or only at the end/interruption).

Nodes to interrupt immediately before they get executed.

Nodes to Nodes to interrupt immediately after they get executed.

Webhook to call after LangGraph API call is done.

Multitask strategy to use. Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'.

Whether to delete or keep the thread created for a stateless run. Must be one of 'delete' or 'keep'.

How to handle missing thread. Defaults to 'reject'. Must be either 'reject' (raise error if missing), or 'create' (create new thread).

The number of seconds to wait before starting the run. Use to schedule future runs.

Optional custom headers to include with the request.

Optional callback to call when a run is created.

Returns:

The created background run.

background_run = await client.runs.create(
    thread_id="my_thread_id",
    assistant_id="my_assistant_id",
    input={"messages": [{"role": "user", "content": "hello!"}]},
    metadata={"name":"my_run"},
    context={"model_name": "openai"},
    interrupt_before=["node_to_stop_before_1","node_to_stop_before_2"],
    interrupt_after=["node_to_stop_after_1","node_to_stop_after_2"],
    webhook="https://my.fake.webhook.com",
    multitask_strategy="interrupt"
)
print(background_run)

--------------------------------------------------------------------------------

{
    'run_id': 'my_run_id',
    'thread_id': 'my_thread_id',
    'assistant_id': 'my_assistant_id',
    'created_at': '2024-07-25T15:35:42.598503+00:00',
    'updated_at': '2024-07-25T15:35:42.598503+00:00',
    'metadata': {},
    'status': 'pending',
    'kwargs':
        {
            'input':
                {
                    'messages': [
                        {
                            'role': 'user',
                            'content': 'how are you?'
                        }
                    ]
                },
            'config':
                {
                    'metadata':
                        {
                            'created_by': 'system'
                        },
                    'configurable':
                        {
                            'run_id': 'my_run_id',
                            'user_id': None,
                            'graph_id': 'agent',
                            'thread_id': 'my_thread_id',
                            'checkpoint_id': None,
                            'assistant_id': 'my_assistant_id'
                        },
                },
            'context':
                {
                    'model_name': 'openai'
                }
            'webhook': "https://my.fake.webhook.com",
            'temporary': False,
            'stream_mode': ['values'],
            'feedback_keys': None,
            'interrupt_after': ["node_to_stop_after_1","node_to_stop_after_2"],
            'interrupt_before': ["node_to_stop_before_1","node_to_stop_before_2"]
        },
    'multitask_strategy': 'interrupt'
}

Create a batch of stateless background runs.

wait(
    thread_id: str | None,
    assistant_id: str,
    *,
    input: dict | None = None,
    command: Command | None = None,
    metadata: dict | None = None,
    config: Config | None = None,
    context: Context | None = None,
    checkpoint: Checkpoint | None = None,
    checkpoint_id: str | None = None,
    checkpoint_during: bool | None = None,
    interrupt_before: All | Sequence[str] | None = None,
    interrupt_after: All | Sequence[str] | None = None,
    webhook: str | None = None,
    on_disconnect: DisconnectMode | None = None,
    on_completion: OnCompletionBehavior | None = None,
    multitask_strategy: MultitaskStrategy | None = None,
    if_not_exists: IfNotExists | None = None,
    after_seconds: int | None = None,
    raise_error: bool = True,
    headers: dict[str, str] | None = None,
    on_run_created: (
        Callable[[RunCreateMetadata], None] | None
    ) = None
) -> list[dict] | dict[str, Any]

Create a run, wait until it finishes and return the final state.

Parameters:

the thread ID to create the run on. If None will create a stateless run.

The assistant ID or graph name to run. If using graph name, will default to first assistant created from that graph.

The input to the graph.

A command to execute. Cannot be combined with input.

Metadata to assign to the run.

The configuration for the assistant.

Static context to add to the assistant.

Supported with langgraph>=0.6.0

The checkpoint to resume from.

Whether to checkpoint during the run (or only at the end/interruption).

Nodes to interrupt immediately before they get executed.

Nodes to Nodes to interrupt immediately after they get executed.

Webhook to call after LangGraph API call is done.

The disconnect mode to use. Must be one of 'cancel' or 'continue'.

Whether to delete or keep the thread created for a stateless run. Must be one of 'delete' or 'keep'.

Multitask strategy to use. Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'.

How to handle missing thread. Defaults to 'reject'. Must be either 'reject' (raise error if missing), or 'create' (create new thread).

The number of seconds to wait before starting the run. Use to schedule future runs.

Optional custom headers to include with the request.

Optional callback to call when a run is created.

Returns:

Union[list[dict], dict[str, Any]]: The output of the run.

client = get_client(url="http://localhost:2024")
final_state_of_run = await client.runs.wait(
    thread_id=None,
    assistant_id="agent",
    input={"messages": [{"role": "user", "content": "how are you?"}]},
    metadata={"name":"my_run"},
    context={"model_name": "anthropic"},
    interrupt_before=["node_to_stop_before_1","node_to_stop_before_2"],
    interrupt_after=["node_to_stop_after_1","node_to_stop_after_2"],
    webhook="https://my.fake.webhook.com",
    multitask_strategy="interrupt"
)
print(final_state_of_run)

-------------------------------------------------------------------------------------------------------------------------------------------

{
    'messages': [
        {
            'content': 'how are you?',
            'additional_kwargs': {},
            'response_metadata': {},
            'type': 'human',
            'name': None,
            'id': 'f51a862c-62fe-4866-863b-b0863e8ad78a',
            'example': False
        },
        {
            'content': "I'm doing well, thanks for asking! I'm an AI assistant created by Anthropic to be helpful, honest, and harmless.",
            'additional_kwargs': {},
            'response_metadata': {},
            'type': 'ai',
            'name': None,
            'id': 'run-bf1cd3c6-768f-4c16-b62d-ba6f17ad8b36',
            'example': False,
            'tool_calls': [],
            'invalid_tool_calls': [],
            'usage_metadata': None
        }
    ]
}

List runs.

Parameters:

The thread ID to list runs for.

The maximum number of results to return.

The number of results to skip.

The status of the run to filter by.

Optional custom headers to include with the request.

Returns:

list[Run]: The runs for the thread.

client = get_client(url="http://localhost:2024")
await client.runs.list(
    thread_id="thread_id",
    limit=5,
    offset=5,
)

Get a run.

Parameters:

The thread ID to get.

The run ID to get.

Optional custom headers to include with the request.

Returns:

Run object.

client = get_client(url="http://localhost:2024")
run = await client.runs.get(
    thread_id="thread_id_to_delete",
    run_id="run_id_to_delete",
)

Get a run.

Parameters:

The thread ID to cancel.

The run ID to cancel.

Whether to wait until run has completed.

Action to take when cancelling the run. Possible values are interrupt or rollback. Default is interrupt.

Optional custom headers to include with the request.

Returns:

None

client = get_client(url="http://localhost:2024")
await client.runs.cancel(
    thread_id="thread_id_to_cancel",
    run_id="run_id_to_cancel",
    wait=True,
    action="interrupt"
)

Block until a run is done. Returns the final state of the thread.

Parameters:

The thread ID to join.

The run ID to join.

Optional custom headers to include with the request.

Returns:

None

client = get_client(url="http://localhost:2024")
result =await client.runs.join(
    thread_id="thread_id_to_join",
    run_id="run_id_to_join"
)

Stream output from a run in real-time, until the run is done. Output is not buffered, so any output produced before this call will not be received here.

Parameters:

The thread ID to join.

The run ID to join.

Whether to cancel the run when the stream is disconnected.

The stream mode(s) to use. Must be a subset of the stream modes passed when creating the run. Background runs default to having the union of all stream modes.

Optional custom headers to include with the request.

Returns:

client = get_client(url="http://localhost:2024")
async for part in client.runs.join_stream(
    thread_id="thread_id_to_join",
    run_id="run_id_to_join",
    stream_mode=["values", "debug"]
):
    print(part)

delete(
    thread_id: str,
    run_id: str,
    *,
    headers: dict[str, str] | None = None
) -> None

Delete a run.

Parameters:

The thread ID to delete.

The run ID to delete.

Optional custom headers to include with the request.

Returns:

None

client = get_client(url="http://localhost:2024")
await client.runs.delete(
    thread_id="thread_id_to_delete",
    run_id="run_id_to_delete"
)

Client for managing recurrent runs (cron jobs) in LangGraph.

A run is a single invocation of an assistant with optional input, config, and context. This client allows scheduling recurring runs to occur automatically.

client = get_client(url="http://localhost:2024"))
cron_job = await client.crons.create_for_thread(
    thread_id="thread_123",
    assistant_id="asst_456",
    schedule="0 9 * * *",
    input={"message": "Daily update"}
)

Feature Availability

The crons client functionality is not supported on all licenses. Please check the relevant license documentation for the most up-to-date details on feature availability.

Methods:

create_for_thread(
    thread_id: str,
    assistant_id: str,
    *,
    schedule: str,
    input: dict | None = None,
    metadata: dict | None = None,
    config: Config | None = None,
    context: Context | None = None,
    checkpoint_during: bool | None = None,
    interrupt_before: All | list[str] | None = None,
    interrupt_after: All | list[str] | None = None,
    webhook: str | None = None,
    multitask_strategy: str | None = None,
    headers: dict[str, str] | None = None
) -> Run

Create a cron job for a thread.

Parameters:

the thread ID to run the cron job on.

The assistant ID or graph name to use for the cron job. If using graph name, will default to first assistant created from that graph.

The cron schedule to execute this job on.

The input to the graph.

Metadata to assign to the cron job runs.

The configuration for the assistant.

Static context to add to the assistant.

Supported with langgraph>=0.6.0

Whether to checkpoint during the run (or only at the end/interruption).

Nodes to interrupt immediately before they get executed.

Nodes to Nodes to interrupt immediately after they get executed.

Webhook to call after LangGraph API call is done.

Multitask strategy to use. Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'.

Optional custom headers to include with the request.

Returns:

The cron run.

client = get_client(url="http://localhost:2024")
cron_run = await client.crons.create_for_thread(
    thread_id="my-thread-id",
    assistant_id="agent",
    schedule="27 15 * * *",
    input={"messages": [{"role": "user", "content": "hello!"}]},
    metadata={"name":"my_run"},
    context={"model_name": "openai"},
    interrupt_before=["node_to_stop_before_1","node_to_stop_before_2"],
    interrupt_after=["node_to_stop_after_1","node_to_stop_after_2"],
    webhook="https://my.fake.webhook.com",
    multitask_strategy="interrupt"
)

create(
    assistant_id: str,
    *,
    schedule: str,
    input: dict | None = None,
    metadata: dict | None = None,
    config: Config | None = None,
    context: Context | None = None,
    checkpoint_during: bool | None = None,
    interrupt_before: All | list[str] | None = None,
    interrupt_after: All | list[str] | None = None,
    webhook: str | None = None,
    multitask_strategy: str | None = None,
    headers: dict[str, str] | None = None
) -> Run

Create a cron run.

Parameters:

The assistant ID or graph name to use for the cron job. If using graph name, will default to first assistant created from that graph.

The cron schedule to execute this job on.

The input to the graph.

Metadata to assign to the cron job runs.

The configuration for the assistant.

Static context to add to the assistant.

Supported with langgraph>=0.6.0

Whether to checkpoint during the run (or only at the end/interruption).

Nodes to interrupt immediately before they get executed.

Nodes to Nodes to interrupt immediately after they get executed.

Webhook to call after LangGraph API call is done.

Multitask strategy to use. Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'.

Optional custom headers to include with the request.

Returns:

The cron run.

client = get_client(url="http://localhost:2024")
cron_run = client.crons.create(
    assistant_id="agent",
    schedule="27 15 * * *",
    input={"messages": [{"role": "user", "content": "hello!"}]},
    metadata={"name":"my_run"},
    context={"model_name": "openai"},
    interrupt_before=["node_to_stop_before_1","node_to_stop_before_2"],
    interrupt_after=["node_to_stop_after_1","node_to_stop_after_2"],
    webhook="https://my.fake.webhook.com",
    multitask_strategy="interrupt"
)

delete(
    cron_id: str, headers: dict[str, str] | None = None
) -> None

Delete a cron.

Parameters:

The cron ID to delete.

Optional custom headers to include with the request.

Returns:

None

client = get_client(url="http://localhost:2024")
await client.crons.delete(
    cron_id="cron_to_delete"
)

Get a list of cron jobs.

Parameters:

The assistant ID or graph name to search for.

the thread ID to search for.

The maximum number of results to return.

The number of results to skip.

Optional custom headers to include with the request.

Returns:

list[Cron]: The list of cron jobs returned by the search,

client = get_client(url="http://localhost:2024")
cron_jobs = await client.crons.search(
    assistant_id="my_assistant_id",
    thread_id="my_thread_id",
    limit=5,
    offset=5,
)
print(cron_jobs)

----------------------------------------------------------

[
    {
        'cron_id': '1ef3cefa-4c09-6926-96d0-3dc97fd5e39b',
        'assistant_id': 'my_assistant_id',
        'thread_id': 'my_thread_id',
        'user_id': None,
        'payload':
            {
                'input': {'start_time': ''},
                'schedule': '4 * * * *',
                'assistant_id': 'my_assistant_id'
            },
        'schedule': '4 * * * *',
        'next_run_date': '2024-07-25T17:04:00+00:00',
        'end_time': None,
        'created_at': '2024-07-08T06:02:23.073257+00:00',
        'updated_at': '2024-07-08T06:02:23.073257+00:00'
    }
]

Client for interacting with the graph's shared storage.

The Store provides a key-value storage system for persisting data across graph executions, allowing for stateful operations and data sharing across threads.

client = get_client(url="http://localhost:2024")
await client.store.put_item(["users", "user123"], "mem-123451342", {"name": "Alice", "score": 100})

Methods:

Store or update an item.

Parameters:

A list of strings representing the namespace path.

The unique identifier for the item within the namespace.

A dictionary containing the item's data.

Controls search indexing - None (use defaults), False (disable), or list of field paths to index.

Optional time-to-live in minutes for the item, or None for no expiration.

Optional custom headers to include with the request.

Returns:

None

client = get_client(url="http://localhost:2024")
await client.store.put_item(
    ["documents", "user123"],
    key="item456",
    value={"title": "My Document", "content": "Hello World"}
)

Retrieve a single item.

Parameters:

The unique identifier for the item.

Optional list of strings representing the namespace path.

Whether to refresh the TTL on this read operation. If None, uses the store's default behavior.

Returns:

The retrieved item.

Optional custom headers to include with the request.

client = get_client(url="http://localhost:2024")
item = await client.store.get_item(
    ["documents", "user123"],
    key="item456",
)
print(item)

----------------------------------------------------------------

{
    'namespace': ['documents', 'user123'],
    'key': 'item456',
    'value': {'title': 'My Document', 'content': 'Hello World'},
    'created_at': '2024-07-30T12:00:00Z',
    'updated_at': '2024-07-30T12:00:00Z'
}

Delete an item.

Parameters:

The unique identifier for the item.

Optional list of strings representing the namespace path.

Optional custom headers to include with the request.

Returns:

None

client = get_client(url="http://localhost:2024")
await client.store.delete_item(
    ["documents", "user123"],
    key="item456",
)

search_items(
    namespace_prefix: Sequence[str],
    /,
    filter: dict[str, Any] | None = None,
    limit: int = 10,
    offset: int = 0,
    query: str | None = None,
    refresh_ttl: bool | None = None,
    headers: dict[str, str] | None = None,
) -> SearchItemsResponse

Search for items within a namespace prefix.

Parameters:

List of strings representing the namespace prefix.

Optional dictionary of key-value pairs to filter results.

Maximum number of items to return (default is 10).

Number of items to skip before returning results (default is 0).

Optional query for natural language search.

Whether to refresh the TTL on items returned by this search. If None, uses the store's default behavior.

Optional custom headers to include with the request.

Returns:

client = get_client(url="http://localhost:2024")
items = await client.store.search_items(
    ["documents"],
    filter={"author": "John Doe"},
    limit=5,
    offset=0
)
print(items)

----------------------------------------------------------------

{
    "items": [
        {
            "namespace": ["documents", "user123"],
            "key": "item789",
            "value": {
                "title": "Another Document",
                "author": "John Doe"
            },
            "created_at": "2024-07-30T12:00:00Z",
            "updated_at": "2024-07-30T12:00:00Z"
        },
        # ... additional items ...
    ]
}

List namespaces with optional match conditions.

Parameters:

Optional list of strings representing the prefix to filter namespaces.

Optional list of strings representing the suffix to filter namespaces.

Optional integer specifying the maximum depth of namespaces to return.

Maximum number of namespaces to return (default is 100).

Number of namespaces to skip before returning results (default is 0).

Optional custom headers to include with the request.

Returns:

client = get_client(url="http://localhost:2024")
namespaces = await client.store.list_namespaces(
    prefix=["documents"],
    max_depth=3,
    limit=10,
    offset=0
)
print(namespaces)

----------------------------------------------------------------

[
    ["documents", "user123", "reports"],
    ["documents", "user456", "invoices"],
    ...
]

Synchronous client for interacting with the LangGraph API.

This class provides synchronous access to LangGraph API endpoints for managing assistants, threads, runs, cron jobs, and data storage.

client = get_sync_client(url="http://localhost:2024")
assistant = client.assistants.get("asst_123")