WEBDIENSTE/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-08-31 20:29:46 +02:00
Code Issues Packages Projects Releases Wiki Activity

sdk-py: Propagate headers in the async client (#4001)

Browse Source 
Propagate headers in the async client
This commit is contained in:
Eugene Yurtsev
2025-03-24 16:30:40 -04:00
committed by GitHub
parent addb491cfd
commit c598fee7d6
1 changed files with 206 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/sdk-py/langgraph_sdk/client.py
+206 -43
View File
@@ -397,11 +397,14 @@ class AssistantsClient:
def __init__(self, http: HttpClient) -> None:
self.http = http
async def get(self, assistant_id: str) -> Assistant:
async def get(
self, assistant_id: str, *, headers: Optional[dict[str, str]] = None
) -> Assistant:
"""Get an assistant by ID.
Args:
assistant_id: The ID of the assistant to get.
headers: Optional custom headers to include with the request.
Returns:
Assistant: Assistant Object.
@@ -427,16 +430,21 @@ class AssistantsClient:
}
""" # noqa: E501
return await self.http.get(f"/assistants/{assistant_id}")
return await self.http.get(f"/assistants/{assistant_id}", headers=headers)
async def get_graph(
self, assistant_id: str, *, xray: Union[int, bool] = False
self,
assistant_id: str,
*,
xray: Union[int, bool] = False,
headers: Optional[dict[str, str]] = None,
) -> dict[str, list[dict[str, Any]]]:
"""Get the graph of an assistant by ID.
Args:
assistant_id: The ID of the assistant to get the graph of.
xray: 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.
headers: Optional custom headers to include with the request.
Returns:
Graph: The graph information for the assistant in JSON format.
@@ -467,14 +475,17 @@ class AssistantsClient:
""" # noqa: E501
return await self.http.get(
f"/assistants/{assistant_id}/graph", params={"xray": xray}
f"/assistants/{assistant_id}/graph", params={"xray": xray}, headers=headers
)
async def get_schemas(self, assistant_id: str) -> GraphSchema:
async def get_schemas(
self, assistant_id: str, *, headers: Optional[dict[str, str]] = None
) -> GraphSchema:
"""Get the schemas of an assistant by ID.
Args:
assistant_id: The ID of the assistant to get the schema of.
headers: Optional custom headers to include with the request.
Returns:
GraphSchema: The graph schema for the assistant.
@@ -573,15 +584,25 @@ class AssistantsClient:
}
""" # noqa: E501
return await self.http.get(f"/assistants/{assistant_id}/schemas")
return await self.http.get(
f"/assistants/{assistant_id}/schemas", headers=headers
)
async def get_subgraphs(
self, assistant_id: str, namespace: Optional[str] = None, recurse: bool = False
self,
assistant_id: str,
namespace: Optional[str] = None,
recurse: bool = False,
*,
headers: Optional[dict[str, str]] = None,
) -> Subgraphs:
"""Get the schemas of an assistant by ID.
Args:
assistant_id: The ID of the assistant to get the schema of.
namespace: Optional namespace to filter by.
recurse: Whether to recursively get subgraphs.
headers: Optional custom headers to include with the request.
Returns:
Subgraphs: The graph schema for the assistant.
@@ -591,11 +612,13 @@ class AssistantsClient:
return await self.http.get(
f"/assistants/{assistant_id}/subgraphs/{namespace}",
params={"recurse": recurse},
headers=headers,
)
else:
return await self.http.get(
f"/assistants/{assistant_id}/subgraphs",
params={"recurse": recurse},
headers=headers,
)
async def create(
@@ -607,6 +630,7 @@ class AssistantsClient:
assistant_id: Optional[str] = None,
if_exists: Optional[OnConflictBehavior] = None,
name: Optional[str] = None,
headers: Optional[dict[str, str]] = None,
) -> Assistant:
"""Create a new assistant.
@@ -620,6 +644,7 @@ class AssistantsClient:
if_exists: 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).
name: The name of the assistant. Defaults to 'Untitled' under the hood.
headers: Optional custom headers to include with the request.
Returns:
Assistant: The created assistant.
@@ -648,7 +673,7 @@ class AssistantsClient:
payload["if_exists"] = if_exists
if name:
payload["name"] = name
return await self.http.post("/assistants", json=payload)
return await self.http.post("/assistants", json=payload, headers=headers)
async def update(
self,
@@ -658,6 +683,7 @@ class AssistantsClient:
config: Optional[Config] = None,
metadata: Json = None,
name: Optional[str] = None,
headers: Optional[dict[str, str]] = None,
) -> Assistant:
"""Update an assistant.
@@ -670,6 +696,7 @@ class AssistantsClient:
config: Configuration to use for the graph.
metadata: Metadata to merge with existing assistant metadata.
name: The new name for the assistant.
headers: Optional custom headers to include with the request.
Returns:
Assistant: The updated assistant.
@@ -696,16 +723,20 @@ class AssistantsClient:
return await self.http.patch(
f"/assistants/{assistant_id}",
json=payload,
headers=headers,
)
async def delete(
self,
assistant_id: str,
*,
headers: Optional[dict[str, str]] = None,
) -> None:
"""Delete an assistant.
Args:
assistant_id: The assistant ID to delete.
headers: Optional custom headers to include with the request.
Returns:
None
@@ -717,7 +748,7 @@ class AssistantsClient:
)
""" # noqa: E501
await self.http.delete(f"/assistants/{assistant_id}")
await self.http.delete(f"/assistants/{assistant_id}", headers=headers)
async def search(
self,
@@ -726,6 +757,7 @@ class AssistantsClient:
graph_id: Optional[str] = None,
limit: int = 10,
offset: int = 0,
headers: Optional[dict[str, str]] = None,
) -> list[Assistant]:
"""Search for assistants.
@@ -735,6 +767,7 @@ class AssistantsClient:
The graph ID is normally set in your langgraph.json configuration.
limit: The maximum number of results to return.
offset: The number of results to skip.
headers: Optional custom headers to include with the request.
Returns:
list[Assistant]: A list of assistants.
@@ -759,6 +792,7 @@ class AssistantsClient:
return await self.http.post(
"/assistants/search",
json=payload,
headers=headers,
)
async def get_versions(
@@ -767,6 +801,8 @@ class AssistantsClient:
metadata: Json = None,
limit: int = 10,
offset: int = 0,
*,
headers: Optional[dict[str, str]] = None,
) -> list[AssistantVersion]:
"""List all versions of an assistant.
@@ -775,6 +811,7 @@ class AssistantsClient:
metadata: Metadata to filter versions by. Exact match filter for each KV pair.
limit: The maximum number of versions to return.
offset: The number of versions to skip.
headers: Optional custom headers to include with the request.
Returns:
list[AssistantVersion]: A list of assistant versions.
@@ -794,15 +831,22 @@ class AssistantsClient:
if metadata:
payload["metadata"] = metadata
return await self.http.post(
f"/assistants/{assistant_id}/versions", json=payload
f"/assistants/{assistant_id}/versions", json=payload, headers=headers
)
async def set_latest(self, assistant_id: str, version: int) -> Assistant:
async def set_latest(
self,
assistant_id: str,
version: int,
*,
headers: Optional[dict[str, str]] = None,
) -> Assistant:
"""Change the version of an assistant.
Args:
assistant_id: The assistant ID to delete.
version: The version to change to.
headers: Optional custom headers to include with the request.
Returns:
Assistant: Assistant Object.
@@ -818,7 +862,9 @@ class AssistantsClient:
payload: Dict[str, Any] = {"version": version}
return await self.http.post(f"/assistants/{assistant_id}/latest", json=payload)
return await self.http.post(
f"/assistants/{assistant_id}/latest", json=payload, headers=headers
)
class ThreadsClient:
@@ -837,11 +883,14 @@ class ThreadsClient:
def __init__(self, http: HttpClient) -> None:
self.http = http
async def get(self, thread_id: str) -> Thread:
async def get(
self, thread_id: str, *, headers: Optional[dict[str, str]] = None
) -> Thread:
"""Get a thread by ID.
Args:
thread_id: The ID of the thread to get.
headers: Optional custom headers to include with the request.
Returns:
Thread: Thread object.
@@ -864,7 +913,7 @@ class ThreadsClient:
""" # noqa: E501
return await self.http.get(f"/threads/{thread_id}")
return await self.http.get(f"/threads/{thread_id}", headers=headers)
async def create(
self,
@@ -874,6 +923,7 @@ class ThreadsClient:
if_exists: Optional[OnConflictBehavior] = None,
supersteps: Optional[Sequence[dict[str, Sequence[dict[str, Any]]]]] = None,
graph_id: Optional[str] = None,
headers: Optional[dict[str, str]] = None,
) -> Thread:
"""Create a new thread.
@@ -886,6 +936,7 @@ class ThreadsClient:
supersteps: 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.
graph_id: Optional graph ID to associate with the thread.
headers: Optional custom headers to include with the request.
Returns:
Thread: The created thread.
@@ -923,14 +974,21 @@ class ThreadsClient:
for s in supersteps
]
return await self.http.post("/threads", json=payload)
return await self.http.post("/threads", json=payload, headers=headers)
async def update(self, thread_id: str, *, metadata: dict[str, Any]) -> Thread:
async def update(
self,
thread_id: str,
*,
metadata: dict[str, Any],
headers: Optional[dict[str, str]] = None,
) -> Thread:
"""Update a thread.
Args:
thread_id: ID of thread to update.
metadata: Metadata to merge with existing thread metadata.
headers: Optional custom headers to include with the request.
Returns:
Thread: The created thread.
@@ -943,14 +1001,17 @@ class ThreadsClient:
)
""" # noqa: E501
return await self.http.patch(
f"/threads/{thread_id}", json={"metadata": metadata}
f"/threads/{thread_id}", json={"metadata": metadata}, headers=headers
)
async def delete(self, thread_id: str) -> None:
async def delete(
self, thread_id: str, *, headers: Optional[dict[str, str]] = None
) -> None:
"""Delete a thread.
Args:
thread_id: The ID of the thread to delete.
headers: Optional custom headers to include with the request.
Returns:
None
@@ -962,7 +1023,7 @@ class ThreadsClient:
)
""" # noqa: E501
await self.http.delete(f"/threads/{thread_id}")
await self.http.delete(f"/threads/{thread_id}", headers=headers)
async def search(
self,
@@ -972,6 +1033,7 @@ class ThreadsClient:
status: Optional[ThreadStatus] = None,
limit: int = 10,
offset: int = 0,
headers: Optional[dict[str, str]] = None,
) -> list[Thread]:
"""Search for threads.
@@ -982,6 +1044,7 @@ class ThreadsClient:
Must be one of 'idle', 'busy', 'interrupted' or 'error'.
limit: Limit on number of threads to return.
offset: Offset in threads table to start search from.
headers: Optional custom headers to include with the request.
Returns:
list[Thread]: List of the threads matching the search parameters.
@@ -1009,13 +1072,17 @@ class ThreadsClient:
return await self.http.post(
"/threads/search",
json=payload,
headers=headers,
)
async def copy(self, thread_id: str) -> None:
async def copy(
self, thread_id: str, *, headers: Optional[dict[str, str]] = None
) -> None:
"""Copy a thread.
Args:
thread_id: The ID of the thread to copy.
headers: Optional custom headers to include with the request.
Returns:
None
@@ -1027,7 +1094,9 @@ class ThreadsClient:
)
""" # noqa: E501
return await self.http.post(f"/threads/{thread_id}/copy", json=None)
return await self.http.post(
f"/threads/{thread_id}/copy", json=None, headers=headers
)
async def get_state(
self,
@@ -1036,13 +1105,16 @@ class ThreadsClient:
checkpoint_id: Optional[str] = None, # deprecated
*,
subgraphs: bool = False,
headers: Optional[dict[str, str]] = None,
) -> ThreadState:
"""Get the state of a thread.
Args:
thread_id: The ID of the thread to get the state of.
checkpoint: The checkpoint to get the state of.
checkpoint_id: (deprecated) The checkpoint ID to get the state of.
subgraphs: Include subgraphs states.
headers: Optional custom headers to include with the request.
Returns:
ThreadState: the thread of the state.
@@ -1134,16 +1206,19 @@ class ThreadsClient:
return await self.http.post(
f"/threads/{thread_id}/state/checkpoint",
json={"checkpoint": checkpoint, "subgraphs": subgraphs},
headers=headers,
)
elif checkpoint_id:
return await self.http.get(
f"/threads/{thread_id}/state/{checkpoint_id}",
params={"subgraphs": subgraphs},
headers=headers,
)
else:
return await self.http.get(
f"/threads/{thread_id}/state",
params={"subgraphs": subgraphs},
headers=headers,
)
async def update_state(
@@ -1154,6 +1229,7 @@ class ThreadsClient:
as_node: Optional[str] = None,
checkpoint: Optional[Checkpoint] = None,
checkpoint_id: Optional[str] = None, # deprecated
headers: Optional[dict[str, str]] = None,
) -> ThreadUpdateStateResponse:
"""Update the state of a thread.
@@ -1162,6 +1238,8 @@ class ThreadsClient:
values: The values to update the state with.
as_node: Update the state as if this node had just executed.
checkpoint: The checkpoint to update the state of.
checkpoint_id: (deprecated) The checkpoint ID to update the state of.
headers: Optional custom headers to include with the request.
Returns:
ThreadUpdateStateResponse: Response after updating a thread's state.
@@ -1196,7 +1274,9 @@ class ThreadsClient:
payload["checkpoint"] = checkpoint
if as_node:
payload["as_node"] = as_node
return await self.http.post(f"/threads/{thread_id}/state", json=payload)
return await self.http.post(
f"/threads/{thread_id}/state", json=payload, headers=headers
)
async def get_history(
self,
@@ -1206,6 +1286,7 @@ class ThreadsClient:
before: Optional[str | Checkpoint] = None,
metadata: Optional[dict] = None,
checkpoint: Optional[Checkpoint] = None,
headers: Optional[dict[str, str]] = None,
) -> list[ThreadState]:
"""Get the state history of a thread.
@@ -1215,6 +1296,7 @@ class ThreadsClient:
limit: The maximum number of states to return.
before: Return states before this checkpoint.
metadata: Filter states by metadata key-value pairs.
headers: Optional custom headers to include with the request.
Returns:
list[ThreadState]: the state history of the thread.
@@ -1236,7 +1318,9 @@ class ThreadsClient:
payload["metadata"] = metadata
if checkpoint:
payload["checkpoint"] = checkpoint
return await self.http.post(f"/threads/{thread_id}/history", json=payload)
return await self.http.post(
f"/threads/{thread_id}/history", json=payload, headers=headers
)
class RunsClient:
@@ -1276,6 +1360,7 @@ class RunsClient:
multitask_strategy: Optional[MultitaskStrategy] = None,
if_not_exists: Optional[IfNotExists] = None,
after_seconds: Optional[int] = None,
headers: Optional[dict[str, str]] = None,
) -> AsyncIterator[StreamPart]: ...
@overload
@@ -1298,6 +1383,7 @@ class RunsClient:
if_not_exists: Optional[IfNotExists] = None,
webhook: Optional[str] = None,
after_seconds: Optional[int] = None,
headers: Optional[dict[str, str]] = None,
) -> AsyncIterator[StreamPart]: ...
def stream(
@@ -1322,6 +1408,7 @@ class RunsClient:
multitask_strategy: Optional[MultitaskStrategy] = None,
if_not_exists: Optional[IfNotExists] = None,
after_seconds: Optional[int] = None,
headers: Optional[dict[str, str]] = None,
) -> AsyncIterator[StreamPart]:
"""Create a run and stream the results.
@@ -1408,7 +1495,10 @@ class RunsClient:
else "/runs/stream"
)
return self.http.stream(
endpoint, "POST", json={k: v for k, v in payload.items() if v is not None}
endpoint,
"POST",
json={k: v for k, v in payload.items() if v is not None},
headers=headers,
)
@overload
@@ -1429,6 +1519,7 @@ class RunsClient:
on_completion: Optional[OnCompletionBehavior] = None,
if_not_exists: Optional[IfNotExists] = None,
after_seconds: Optional[int] = None,
headers: Optional[dict[str, str]] = None,
) -> Run: ...
@overload
@@ -1451,6 +1542,7 @@ class RunsClient:
multitask_strategy: Optional[MultitaskStrategy] = None,
if_not_exists: Optional[IfNotExists] = None,
after_seconds: Optional[int] = None,
headers: Optional[dict[str, str]] = None,
) -> Run: ...
async def create(
@@ -1473,6 +1565,7 @@ class RunsClient:
if_not_exists: Optional[IfNotExists] = None,
on_completion: Optional[OnCompletionBehavior] = None,
after_seconds: Optional[int] = None,
headers: Optional[dict[str, str]] = None,
) -> Run:
"""Create a background run.
@@ -1499,6 +1592,7 @@ class RunsClient:
Must be either 'reject' (raise error if missing), or 'create' (create new thread).
after_seconds: The number of seconds to wait before starting the run.
Use to schedule future runs.
headers: Optional custom headers to include with the request.
Returns:
Run: The created background run.
@@ -1622,6 +1716,7 @@ class RunsClient:
if_not_exists: Optional[IfNotExists] = None,
after_seconds: Optional[int] = None,
raise_error: bool = True,
headers: Optional[dict[str, str]] = None,
) -> Union[list[dict], dict[str, Any]]: ...
@overload
@@ -1642,6 +1737,7 @@ class RunsClient:
if_not_exists: Optional[IfNotExists] = None,
after_seconds: Optional[int] = None,
raise_error: bool = True,
headers: Optional[dict[str, str]] = None,
) -> Union[list[dict], dict[str, Any]]: ...
async def wait(
@@ -1664,6 +1760,7 @@ class RunsClient:
if_not_exists: Optional[IfNotExists] = None,
after_seconds: Optional[int] = None,
raise_error: bool = True,
headers: Optional[dict[str, str]] = None,
) -> Union[list[dict], dict[str, Any]]:
"""Create a run, wait until it finishes and return the final state.
@@ -1690,6 +1787,7 @@ class RunsClient:
Must be either 'reject' (raise error if missing), or 'create' (create new thread).
after_seconds: The number of seconds to wait before starting the run.
Use to schedule future runs.
headers: Optional custom headers to include with the request.
Returns:
Union[list[dict], dict[str, Any]]: The output of the run.
@@ -1761,7 +1859,9 @@ class RunsClient:
f"/threads/{thread_id}/runs/wait" if thread_id is not None else "/runs/wait"
)
response = await self.http.post(
endpoint, json={k: v for k, v in payload.items() if v is not None}
endpoint,
json={k: v for k, v in payload.items() if v is not None},
headers=headers,
)
if (
raise_error
@@ -1781,6 +1881,7 @@ class RunsClient:
limit: int = 10,
offset: int = 0,
status: Optional[RunStatus] = None,
headers: Optional[dict[str, str]] = None,
) -> List[Run]:
"""List runs.
@@ -1789,6 +1890,7 @@ class RunsClient:
limit: The maximum number of results to return.
offset: The number of results to skip.
status: The status of the run to filter by.
headers: Optional custom headers to include with the request.
Returns:
List[Run]: The runs for the thread.
@@ -1808,14 +1910,19 @@ class RunsClient:
}
if status is not None:
params["status"] = status
return await self.http.get(f"/threads/{thread_id}/runs", params=params)
return await self.http.get(
f"/threads/{thread_id}/runs", params=params, headers=headers
)
async def get(self, thread_id: str, run_id: str) -> Run:
async def get(
self, thread_id: str, run_id: str, *, headers: Optional[dict[str, str]] = None
) -> Run:
"""Get a run.
Args:
thread_id: The thread ID to get.
run_id: The run ID to get.
headers: Optional custom headers to include with the request.
Returns:
Run: Run object.
@@ -1829,7 +1936,9 @@ class RunsClient:
""" # noqa: E501
return await self.http.get(f"/threads/{thread_id}/runs/{run_id}")
return await self.http.get(
f"/threads/{thread_id}/runs/{run_id}", headers=headers
)
async def cancel(
self,
@@ -1838,6 +1947,7 @@ class RunsClient:
*,
wait: bool = False,
action: CancelAction = "interrupt",
headers: Optional[dict[str, str]] = None,
) -> None:
"""Get a run.
@@ -1847,6 +1957,7 @@ class RunsClient:
wait: Whether to wait until run has completed.
action: Action to take when cancelling the run. Possible values
are `interrupt` or `rollback`. Default is `interrupt`.
headers: Optional custom headers to include with the request.
Returns:
None
@@ -1864,14 +1975,18 @@ class RunsClient:
return await self.http.post(
f"/threads/{thread_id}/runs/{run_id}/cancel?wait={1 if wait else 0}&action={action}",
json=None,
headers=headers,
)
async def join(self, thread_id: str, run_id: str) -> dict:
async def join(
self, thread_id: str, run_id: str, *, headers: Optional[dict[str, str]] = None
) -> dict:
"""Block until a run is done. Returns the final state of the thread.
Args:
thread_id: The thread ID to join.
run_id: The run ID to join.
headers: Optional custom headers to include with the request.
Returns:
None
@@ -1884,7 +1999,9 @@ class RunsClient:
)
""" # noqa: E501
return await self.http.get(f"/threads/{thread_id}/runs/{run_id}/join")
return await self.http.get(
f"/threads/{thread_id}/runs/{run_id}/join", headers=headers
)
def join_stream(
self,
@@ -1893,6 +2010,7 @@ class RunsClient:
*,
cancel_on_disconnect: bool = False,
stream_mode: Optional[Union[StreamMode, Sequence[StreamMode]]] = None,
headers: Optional[dict[str, str]] = None,
) -> AsyncIterator[StreamPart]:
"""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
@@ -1905,6 +2023,7 @@ class RunsClient:
stream_mode: 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.
headers: Optional custom headers to include with the request.
Returns:
None
@@ -1925,14 +2044,18 @@ class RunsClient:
"cancel_on_disconnect": cancel_on_disconnect,
"stream_mode": stream_mode,
},
headers=headers,
)
async def delete(self, thread_id: str, run_id: str) -> None:
async def delete(
self, thread_id: str, run_id: str, *, headers: Optional[dict[str, str]] = None
) -> None:
"""Delete a run.
Args:
thread_id: The thread ID to delete.
run_id: The run ID to delete.
headers: Optional custom headers to include with the request.
Returns:
None
@@ -1945,7 +2068,7 @@ class RunsClient:
)
""" # noqa: E501
await self.http.delete(f"/threads/{thread_id}/runs/{run_id}")
await self.http.delete(f"/threads/{thread_id}/runs/{run_id}", headers=headers)
class CronClient:
@@ -1981,6 +2104,7 @@ class CronClient:
interrupt_after: Optional[Union[All, list[str]]] = None,
webhook: Optional[str] = None,
multitask_strategy: Optional[str] = None,
headers: Optional[dict[str, str]] = None,
) -> Run:
"""Create a cron job for a thread.
@@ -1999,6 +2123,7 @@ class CronClient:
webhook: Webhook to call after LangGraph API call is done.
multitask_strategy: Multitask strategy to use.
Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'.
headers: Optional custom headers to include with the request.
Returns:
Run: The cron run.
@@ -2032,7 +2157,9 @@ class CronClient:
if multitask_strategy:
payload["multitask_strategy"] = multitask_strategy
payload = {k: v for k, v in payload.items() if v is not None}
return await self.http.post(f"/threads/{thread_id}/runs/crons", json=payload)
return await self.http.post(
f"/threads/{thread_id}/runs/crons", json=payload, headers=headers
)
async def create(
self,
@@ -2046,6 +2173,7 @@ class CronClient:
interrupt_after: Optional[Union[All, list[str]]] = None,
webhook: Optional[str] = None,
multitask_strategy: Optional[str] = None,
headers: Optional[dict[str, str]] = None,
) -> Run:
"""Create a cron run.
@@ -2061,6 +2189,7 @@ class CronClient:
webhook: Webhook to call after LangGraph API call is done.
multitask_strategy: Multitask strategy to use.
Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'.
headers: Optional custom headers to include with the request.
Returns:
Run: The cron run.
@@ -2093,13 +2222,18 @@ class CronClient:
if multitask_strategy:
payload["multitask_strategy"] = multitask_strategy
payload = {k: v for k, v in payload.items() if v is not None}
return await self.http.post("/runs/crons", json=payload)
return await self.http.post("/runs/crons", json=payload, headers=headers)
async def delete(self, cron_id: str) -> None:
async def delete(
self,
cron_id: str,
headers: Optional[dict[str, str]] = None,
) -> None:
"""Delete a cron.
Args:
cron_id: The cron ID to delete.
headers: Optional custom headers to include with the request.
Returns:
None
@@ -2111,7 +2245,7 @@ class CronClient:
)
""" # noqa: E501
await self.http.delete(f"/runs/crons/{cron_id}")
await self.http.delete(f"/runs/crons/{cron_id}", headers=headers)
async def search(
self,
@@ -2120,6 +2254,7 @@ class CronClient:
thread_id: Optional[str] = None,
limit: int = 10,
offset: int = 0,
headers: Optional[dict[str, str]] = None,
) -> list[Cron]:
"""Get a list of cron jobs.
@@ -2128,6 +2263,7 @@ class CronClient:
thread_id: the thread ID to search for.
limit: The maximum number of results to return.
offset: The number of results to skip.
headers: Optional custom headers to include with the request.
Returns:
list[Cron]: The list of cron jobs returned by the search,
@@ -2172,7 +2308,7 @@ class CronClient:
"offset": offset,
}
payload = {k: v for k, v in payload.items() if v is not None}
return await self.http.post("/runs/crons/search", json=payload)
return await self.http.post("/runs/crons/search", json=payload, headers=headers)
class StoreClient:
@@ -2198,6 +2334,7 @@ class StoreClient:
value: dict[str, Any],
index: Optional[Union[Literal[False], list[str]]] = None,
ttl: Optional[int] = None,
headers: Optional[dict[str, str]] = None,
) -> None:
"""Store or update an item.
@@ -2207,6 +2344,7 @@ class StoreClient:
value: A dictionary containing the item's data.
index: Controls search indexing - None (use defaults), False (disable), or list of field paths to index.
ttl: Optional time-to-live in minutes for the item, or None for no expiration.
headers: Optional custom headers to include with the request.
Returns:
None
@@ -2231,7 +2369,9 @@ class StoreClient:
"index": index,
"ttl": ttl,
}
await self.http.put("/store/items", json=_provided_vals(payload))
await self.http.put(
"/store/items", json=_provided_vals(payload), headers=headers
)
async def get_item(
self,
@@ -2240,6 +2380,7 @@ class StoreClient:
key: str,
*,
refresh_ttl: Optional[bool] = None,
headers: Optional[dict[str, str]] = None,
) -> Item:
"""Retrieve a single item.
@@ -2250,6 +2391,7 @@ class StoreClient:
Returns:
Item: The retrieved item.
headers: Optional custom headers to include with the request.
Example Usage:
@@ -2277,14 +2419,21 @@ class StoreClient:
params = {"namespace": ".".join(namespace), "key": key}
if refresh_ttl is not None:
params["refresh_ttl"] = refresh_ttl
return await self.http.get("/store/items", params=params)
return await self.http.get("/store/items", params=params, headers=headers)
async def delete_item(self, namespace: Sequence[str], /, key: str) -> None:
async def delete_item(
self,
namespace: Sequence[str],
/,
key: str,
headers: Optional[dict[str, str]] = None,
) -> None:
"""Delete an item.
Args:
key: The unique identifier for the item.
namespace: Optional list of strings representing the namespace path.
headers: Optional custom headers to include with the request.
Returns:
None
@@ -2297,7 +2446,9 @@ class StoreClient:
)
"""
await self.http.delete(
"/store/items", json={"namespace": namespace, "key": key}
"/store/items",
json={"namespace": namespace, "key": key},
headers=headers,
)
async def search_items(
@@ -2309,6 +2460,7 @@ class StoreClient:
offset: int = 0,
query: Optional[str] = None,
refresh_ttl: Optional[bool] = None,
headers: Optional[dict[str, str]] = None,
) -> SearchItemsResponse:
"""Search for items within a namespace prefix.
@@ -2319,6 +2471,7 @@ class StoreClient:
offset: Number of items to skip before returning results (default is 0).
query: Optional query for natural language search.
refresh_ttl: Whether to refresh the TTL on items returned by this search. If None, uses the store's default behavior.
headers: Optional custom headers to include with the request.
Returns:
List[Item]: A list of items matching the search criteria.
@@ -2360,7 +2513,11 @@ class StoreClient:
"refresh_ttl": refresh_ttl,
}
return await self.http.post("/store/items/search", json=_provided_vals(payload))
return await self.http.post(
"/store/items/search",
json=_provided_vals(payload),
headers=headers,
)
async def list_namespaces(
self,
@@ -2369,6 +2526,7 @@ class StoreClient:
max_depth: Optional[int] = None,
limit: int = 100,
offset: int = 0,
headers: Optional[dict[str, str]] = None,
) -> ListNamespaceResponse:
"""List namespaces with optional match conditions.
@@ -2378,6 +2536,7 @@ class StoreClient:
max_depth: Optional integer specifying the maximum depth of namespaces to return.
limit: Maximum number of namespaces to return (default is 100).
offset: Number of namespaces to skip before returning results (default is 0).
headers: Optional custom headers to include with the request.
Returns:
List[List[str]]: A list of namespaces matching the criteria.
@@ -2407,7 +2566,11 @@ class StoreClient:
"limit": limit,
"offset": offset,
}
return await self.http.post("/store/namespaces", json=_provided_vals(payload))
return await self.http.post(
"/store/namespaces",
json=_provided_vals(payload),
headers=headers,
)
def get_sync_client(