KitaruSyncClient

Synchronous Kitaru client.

Attributes

attributeapiKitaruAPIClient

API client with one method per endpoint.

Functions

func__init__(self, api_client=None) -> None

Initialize the client.

paramself

paramapi_clientKitaruAPIClient | None

= None

API client used to send requests.

Returns

None

funcclose(self) -> None

Close the underlying client and stop the event loop thread.

paramself

Returns

None

funcget_agent(self, agent) -> AgentResponse

Get an agent by id or name.

paramself

paramagentuuid.UUID | str

Id or name of the agent.

Returns

kitaru.api_models.v1.agent.AgentResponse

Stored agent.

funclist_agents(self) -> Iterator[AgentResponse]

Iterate over all agents.

paramself

Returns

collections.abc.Iterator

Iterator over every agent.

funcget_session(self, session_id) -> SessionDetailResponse

Get a session by id.

paramself

paramsession_iduuid.UUID

Id of the session.

Returns

kitaru.api_models.v1.session.SessionDetailResponse

Stored session.

funclist_sessions(self, agent=None) -> Iterator[SessionResponse]

Iterate over sessions, optionally scoped to one agent.

paramself

paramagentuuid.UUID | str | None

= None

Id or name of the agent to scope to.

Returns

collections.abc.Iterator

Iterator over every matching session.

funclist_session_nodes(self, session_id) -> Iterator[SessionNodeResponse]

Iterate over the nodes of a session in index order.

paramself

paramsession_iduuid.UUID

Id of the session.

Returns

collections.abc.Iterator

Iterator over every node.

funcreplay

(self, session_id, evaluators, agent_version_id=None, override=None, tool_policy=None, baseline_evaluation_mode=BaselineEvaluationMode.IF_MISSING, wait=True, timeout=None) -> ReplayResponse

Replay a session and optionally wait for it to finish.

paramself

paramsession_iduuid.UUID

Id of the session to replay.

paramevaluatorslist[EvaluatorConfig]

Evaluators run against the result session.

paramagent_version_iduuid.UUID | None

= None

Agent version to replay with, the session's recorded version when unset.

paramoverrideReplayOverride | None

= None

Override to apply.

paramtool_policyToolPolicy | None

= None

Tool policy to apply.

parambaseline_evaluation_modeBaselineEvaluationMode

= BaselineEvaluationMode.IF_MISSING

How to score the baseline session.

paramwaitbool

= True

Whether to wait for the replay to reach a terminal status.

paramtimeoutfloat | None

= None

Seconds to wait before giving up.

Returns

kitaru.api_models.v1.replay.ReplayResponse

Replay, in a terminal status when waited for.

funcget_replay(self, replay_id) -> ReplayResponse

Get a replay by id.

paramself

paramreplay_iduuid.UUID

Id of the replay.

Returns

kitaru.api_models.v1.replay.ReplayResponse

Stored replay.

funcwait_for_replay(self, replay_id, timeout=None, poll_interval=2.0) -> ReplayResponse

Wait for a replay to reach a terminal status.

paramself

paramreplay_iduuid.UUID

Id of the replay.

paramtimeoutfloat | None

= None

Seconds to wait before giving up.

parampoll_intervalfloat

= 2.0

Seconds between status checks.

Returns

kitaru.api_models.v1.replay.ReplayResponse

Replay in a terminal status.

funcget_experiment(self, experiment) -> ExperimentResponse

Get an experiment by id or name.

paramself

paramexperimentuuid.UUID | str

Id or name of the experiment.

Returns

kitaru.api_models.v1.experiment.ExperimentResponse

Stored experiment.

funclist_experiments(self) -> Iterator[ExperimentResponse]

Iterate over all experiments.

paramself

Returns

collections.abc.Iterator

Iterator over every experiment.

funcrun_experiment

(self, experiment, cohort_version_id, agent_version_id, baseline_evaluation_mode=BaselineEvaluationMode.IF_MISSING, wait=True, timeout=None) -> ExperimentRunResponse

Start an experiment run and optionally wait for it to finish.

paramself

paramexperimentuuid.UUID | str

Id or name of the experiment.

paramcohort_version_iduuid.UUID

Cohort version whose sessions are replayed.

paramagent_version_iduuid.UUID

Agent version to replay with.

parambaseline_evaluation_modeBaselineEvaluationMode

= BaselineEvaluationMode.IF_MISSING

How to score each baseline session.

paramwaitbool

= True

Whether to wait for the run to reach a terminal status.

paramtimeoutfloat | None

= None

Seconds to wait before giving up.

Returns

kitaru.api_models.v1.experiment_run.ExperimentRunResponse

Experiment run, in a terminal status when waited for.

funcget_experiment_run(self, run_id) -> ExperimentRunResponse

Get an experiment run by id.

paramself

paramrun_iduuid.UUID

Id of the experiment run.

Returns

kitaru.api_models.v1.experiment_run.ExperimentRunResponse

Stored experiment run.

funcwait_for_experiment_run(self, run_id, timeout=None, poll_interval=2.0) -> ExperimentRunResponse

Wait for an experiment run to reach a terminal status.

paramself

paramrun_iduuid.UUID

Id of the experiment run.

paramtimeoutfloat | None

= None

Seconds to wait before giving up.

parampoll_intervalfloat

= 2.0

Seconds between status checks.

Returns

kitaru.api_models.v1.experiment_run.ExperimentRunResponse

Experiment run in a terminal status.

Attributes

Functions

On this page