Asynchronous client
Asynchronous client
Use AsyncTypeSafeClient to ask questions, list models, and configure asynchronous TypeSafe API requests.
typesafe_sdk.AsyncTypeSafeClient
AsyncTypeSafeClient(
*,
api_key: str | None = None,
model: str | None = None,
retry: RetryPolicy | None = None,
timeout: float
| httpx2.Timeout
| None = None,
headers: Mapping[str, str] | None = None,
transport: httpx2.AsyncBaseTransport
| None = None,
http_client: httpx2.AsyncClient
| None = None,
base_url: str | None = None,
)
Create an asynchronous HTTP client for TypeSafe AI API.
Explicit options take precedence over environment variables; empty or whitespace-only environment values are ignored.
Parameters:
-
api_key(str | None, default:None) –Required API key; may be set via the
TYPESAFE_API_KEYenvironment variable. Leading and trailing whitespace is stripped. Empty keys, internal whitespace, control characters, and non-ASCII characters are rejected. -
model(str | None, default:None) –Model name; may be set via the
TYPESAFE_DEFAULT_MODELenvironment variable. -
retry(RetryPolicy | None, default:None) –A
RetryPolicycontrolling retry behavior; seeRetryPolicyfor the available options and their defaults. PassRetryPolicy(max_retries=0)to disable retries. -
timeout(float | httpx2.Timeout | None, default:None) –Timeout for HTTP operations. Inherits
http_client.timeoutwhen supplied, otherwise the SDK default. -
headers(Mapping[str, str] | None, default:None) –Additional request headers to set.
-
transport(httpx2.AsyncBaseTransport | None, default:None) –Optional custom HTTP transport, closed when this SDK client closes.
-
http_client(httpx2.AsyncClient | None, default:None) –Optional
httpx2.AsyncClient; mutually exclusive withtransport. Closed when this SDK client closes. -
base_url(str | None, default:None) –API root; may be set via the
TYPESAFE_BASE_URLenvironment variable.
Raises:
-
The API key is missing or invalid, or the timeout is invalid.
-
Both
transportandhttp_clientare supplied.
Examples:
import asyncio
from typesafe_sdk import AsyncTypeSafeClient, Choice, Noul
async def main() -> None:
async with AsyncTypeSafeClient() as client:
result = await client.system_one(
state="I was charged twice. Please help.",
questions={
"billing": Noul(instructions="Is this about billing?"),
"tone": Choice(
instructions="What is the tone?",
criteria={"calm": None, "angry": None},
),
},
)
assert 0 <= result.nouls["billing"].noul <= 1
assert result.choices["tone"].choice in {"calm", "angry"}
asyncio.run(main())
models
cached property
models: AsyncModels
An accessor for the Models API resource.
Examples:
async def main() -> None:
async with AsyncTypeSafeClient() as client:
models = await client.models.list()
system_one
async
system_one(
state: JSONContent,
questions: Mapping[str, Question],
*,
model: str | None = None,
retry: RetryPolicy | None = None,
timeout: float
| httpx2.Timeout
| None = None,
extra_headers: Mapping[str, str]
| None = None,
extra_body: Mapping[str, JSONValue | None]
| None = None,
response_model: type[ResponseT]
| None = None,
) -> SystemOneResponse | ResponseT
system_one(
state: JSONContent,
questions: Mapping[str, Question],
*,
model: str | None = None,
retry: RetryPolicy | None = None,
timeout: float
| httpx2.Timeout
| None = None,
extra_headers: Mapping[str, str]
| None = None,
extra_body: Mapping[str, JSONValue | None]
| None = None,
response_model: None = None,
) -> SystemOneResponse
system_one(
state: JSONContent,
questions: Mapping[str, Question],
*,
model: str | None = None,
retry: RetryPolicy | None = None,
timeout: float
| httpx2.Timeout
| None = None,
extra_headers: Mapping[str, str]
| None = None,
extra_body: Mapping[str, JSONValue | None]
| None = None,
response_model: type[ResponseT],
) -> ResponseT
Answer named questions about text or structured state.
See System One for details.
Parameters:
-
state(JSONContent) –Text, a JSON object, or an array to evaluate. See state for details.
-
questions(Mapping[str, Question]) –Nonempty mapping of names to question objects or raw dictionaries.
-
model(str | None, default:None) –Model override;
Noneinherits the client default. -
retry(RetryPolicy | None, default:None) –An optional retry policy to override the client-level value for this call only.
-
timeout(float | httpx2.Timeout | None, default:None) –An optional timeout for http operations to override the client-level value for this call only, in seconds.
-
extra_headers(Mapping[str, str] | None, default:None) –Additional request headers to set.
-
extra_body(Mapping[str, JSONValue | None] | None, default:None) –Additional top-level request-body fields, shallow-merged over the body after
state,model, andquestionsare set. Merging is last-write-wins: a key that collides withstate,model, orquestionsoverrides it, and object values are replaced rather than deep-merged. -
response_model(type[ResponseT] | None, default:None) –Optional Pydantic
BaseModeltype describing the JSON response body, including any nested answer models.
Returns:
-
SystemOneResponse | ResponseT–An instance of
response_model, orSystemOneResponsewith answers keyed by question -
SystemOneResponse | ResponseT–name and model and token usage details when no custom model is supplied.
Raises:
-
Questions are empty or a score question’s criteria list is empty.
-
The server returns an unsuccessful HTTP response after any retries.
-
The request cannot connect or times out after any retries.
-
TypeSafeAPIResponseValidationError–The response body does not match the response model.
Examples:
Create questions with named arguments:
async def main() -> None:
async with AsyncTypeSafeClient() as client:
result = await client.system_one(
state="I was charged twice. Please help.",
questions={
"billing": Noul(instructions="Is this about billing?"),
"tone": Choice(
instructions="What is the tone?",
criteria={"calm": None, "angry": None},
),
},
)
assert 0 <= result.nouls["billing"].noul <= 1
assert result.choices["tone"].choice in {"calm", "angry"}
Pass questions as dictionaries:
async def main() -> None:
async with AsyncTypeSafeClient() as client:
result = await client.system_one(
state={"message": "I was charged twice. Please help."},
questions={
"billing": {"type": "noul", "instructions": "Is this about billing?"},
"tone": {
"type": "choice",
"instructions": "What is the tone?",
"criteria": {"calm": None, "angry": None},
},
},
)
assert 0 <= result.nouls["billing"].noul <= 1
assert result.choices["tone"].choice in {"calm", "angry"}
aclose
async
aclose() -> None
Release network resources and close the underlying HTTP client, including a supplied one.
Models resource
Reached through AsyncTypeSafeClient.models.
typesafe_sdk.AsyncModels
Access to the models available to the account, reached through AsyncTypeSafeClient.models.
list
async
list(
*,
retry: RetryPolicy | None = None,
timeout: float
| httpx2.Timeout
| None = None,
extra_headers: Mapping[str, str]
| None = None,
) -> ListModelsResponse
List the models available to the account.
Parameters:
-
retry(RetryPolicy | None, default:None) –An optional retry policy to override the client-level value for this call only.
-
timeout(float | httpx2.Timeout | None, default:None) –Per-operation timeout override;
Noneinherits the client setting. -
extra_headers(Mapping[str, str] | None, default:None) –Overrides for additional request headers; authentication, SDK identification, and
Acceptremain protected.
Returns:
-
A
ListModelsResponsewhosemodelsholds each model’s name, description, -
and release date.
Raises:
-
The server returns an unsuccessful HTTP response after any retries.
-
The request cannot connect or times out after any retries.
Examples:
from typesafe_sdk import AsyncTypeSafeClient
async def main() -> None:
async with AsyncTypeSafeClient() as client:
models = await client.models.list()