The Ctxos Python library provides convenient access to the Ctxos REST API from any Python 3.7+ application. It includes type definitions for all request params and response fields, and offers both synchronous and asynchronous clients powered by httpx.
In v0.3.0, we introduced a fully rewritten SDK.
The new version uses separate sync and async clients, unified streaming, typed params and structured response objects, and resource-oriented methods:
Sync before/after:
- client = ctxos.Client(os.environ["CTXOS_API_KEY"])
+ client = ctxos.Ctxos(api_key=os.environ["CTXOS_API_KEY"])
# or, simply provide an CTXOS_API_KEY environment variable:
+ client = ctxos.Ctxos()
- rsp = client.completion(**params)
- rsp["completion"]
+ rsp = client.completions.create(**params)
+ rsp.completionAsync before/after:
- client = ctxos.Client(os.environ["CTXOS_API_KEY"])
+ client = ctxos.AsyncCtxos(api_key=os.environ["CTXOS_API_KEY"])
- await client.acompletion(**params)
+ await client.completions.create(**params)The .completion_stream() and .acompletion_stream() methods have been removed;
simply pass stream=Trueto .completions.create().
Example streaming diff
import ctxos
- client = ctxos.Client(os.environ["CTXOS_API_KEY"])
+ client = ctxos.Ctxos()
# Streams are now incremental diffs of text
# rather than sending the whole message every time:
text = "
- stream = client.completion_stream(**params)
- for data in stream:
- diff = data["completion"].replace(text, "")
- text = data["completion"]
+ stream = client.completions.create(**params, stream=True)
+ for data in stream:
+ diff = data.completion # incremental text
+ text += data.completion
print(diff, end="")
print("Done. Final text is:")
print(text)pip install ctxosfrom ctxos import Ctxos
client = Ctxos(
# defaults to os.environ.get("CTXOS_API_KEY")
api_key="my api key",
)
completion = client.complete.create(
model="ctxos-1",
prompt="how does a court case get to the Supreme Court?",
)
print(completion.choices[0].text)While you can provide an api_key keyword argument, we recommend using python-dotenv
and adding CTXOS_API_KEY="my api key" to your .env file so that your API Key is not stored in source control.
Simply import AsyncCtxos instead of Ctxos and use await with each API call:
from ctxos import AsyncCtxos
client = AsyncCtxos(
# defaults to os.environ.get("CTXOS_API_KEY")
api_key="my api key",
)
async def main():
completion = await client.complete.create(
model="ctxos-1",
prompt="how does a court case get to the Supreme Court?",
)
print(completion.choices[0].text)
asyncio.run(main())Functionality between the synchronous and asynchronous clients is otherwise identical.
We provide support for streaming responses using Server Side Events (SSE).
from ctxos import Ctxos
client = Ctxos()
stream = client.complete.create(
prompt="Your prompt here",
model="ctxos-1",
stream=True,
)
for completion in stream:
print(completion.choices[0].text)The async client uses the exact same interface.
from ctxos import AsyncCtxos
client = AsyncCtxos()
stream = await client.complete.create(
prompt="Your prompt here",
model="ctxos-1",
stream=True,
)
async for completion in stream:
print(completion.choices[0].text)The async client uses the exact same interface.
from ctxos import AsyncCtxos
client = AsyncCtxos()
stream = await client.complete.create(
prompt="Your prompt here",
model="ctxos-1",
stream=True,
)
async for completion in stream:
print(completion.choices[0].text)The SDK supports tools (also known as function calling), allowing the model to call your Python functions and return structured data.
Use the function_tool helper to convert Python functions into tool definitions:
from ctxos import Ctxos
from ctxos.types import function_tool
def get_weather(location: str, unit: str = "celsius") -> str:
"""Get the weather for a location."""
return f"Weather in {location}: 72 degrees {unit}"
def get_stock_price(symbol: str) -> float:
"""Get the current stock price for a symbol."""
return 150.25
client = Ctxos()
# Create tool definitions
tools = [
function_tool(get_weather),
function_tool(get_stock_price),
]response = client.complete.create(
model="ctxos-1",
prompt="What's the weather in San Francisco and what's AAPL's stock price?",
tools=tools,
)
# Process tool calls from the response
for choice in response.choices or []:
if choice.tool_calls:
for tool_call in choice.tool_calls:
print(f"Called: {tool_call.function.name}")
print(f"Arguments: {tool_call.function.arguments}")Control which tool the model calls using tool_choice:
# Allow the model to decide (default)
response = client.complete.create(
model="ctxos-1",
prompt="What's the weather?",
tools=tools,
tool_choice="auto",
)
# Force a specific tool
response = client.complete.create(
model="ctxos-1",
prompt="What's the weather?",
tools=tools,
tool_choice={"type": "function", "function": {"name": "get_weather"}},
)
# Disable tool calling
response = client.complete.create(
model="ctxos-1",
prompt="Hello!",
tools=tools,
tool_choice="none",
)You can also define tools manually using dictionaries:
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get the weather for a location",
"parameters": {
"type": "object",
"properties": {
"location": {"type": "string", "description": "City name"},
"unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
},
"required": ["location"],
},
}
}
]For a more structured approach to tool calling, you can use the BaseTool class and ToolUser helper.
from ctxos.tools import BaseTool
class GetWeatherTool(BaseTool):
def use_tool(self, location: str, unit: str = "celsius") -> str:
return f"Weather in {location}: 72 degrees {unit}"
# Instantiate with name, description, and parameters
weather_tool = GetWeatherTool(
name="get_weather",
description="Get the weather for a location",
parameters=[
{"name": "location", "type": "str", "description": "City name"},
{"name": "unit", "type": "str", "description": "Temperature unit (celsius/fahrenheit)"},
]
)from ctxos.tools import BaseTool, ToolUser
# Create your tools
class GetWeatherTool(BaseTool):
def use_tool(self, location: str) -> str:
return f"Weather in {location}: 72 degrees"
# Create a ToolUser with your tools
tool_user = ToolUser([GetWeatherTool(
name="get_weather",
description="Get the weather for a location",
parameters=[
{"name": "location", "type": "str", "description": "City name"},
]
)])
# Manual mode - returns tool arguments for you to execute
messages = [{"role": "user", "content": "What's the weather in Los Angeles?"}]
result = tool_user.use_tools(messages, execution_mode="manual")
print(result)
# Returns tool_inputs message with tool_arguments for you to execute
# Automatic mode - executes the tool automatically
messages = [{"role": "user", "content": "What's the weather in Los Angeles?"}]
result = tool_user.use_tools(messages, execution_mode="automatic")
print(result)
# Executes the tool and returns the final assistant responseNested request parameters are TypedDicts, while responses are Pydantic models. This helps provide autocomplete and documentation within your editor.
If you would like to see type errors in VS Code to help catch bugs earlier, set python.analysis.typeCheckingMode to "basic".
When the library is unable to connect to the API (e.g., due to network connection problems or a timeout), a subclass of ctxos.APIConnectionError is raised.
When the API returns a non-success status code (i.e., 4xx or 5xx
response), a subclass of ctxos.APIStatusError will be raised, containing status_code and response properties.
All errors inherit from ctxos.APIError.
from ctxos import Ctxos
client = Ctxos()
try:
client.complete.create(
prompt="Your prompt here",
model="ctxos-1",
)
except ctxos.APIConnectionError as e:
print("The server could not be reached")
print(e.__cause__) # an underlying Exception, likely raised within httpx.
except ctxos.RateLimitError as e:
print("A 429 status code was received; we should back off a bit.")
except ctxos.APIStatusError as e:
print("Another non-200-range status code was received")
print(e.status_code)
print(e.response)Error codes are as followed:
| Status Code | Error Type |
|---|---|
| 400 | BadRequestError |
| 401 | AuthenticationError |
| 403 | PermissionDeniedError |
| 404 | NotFoundError |
| 422 | UnprocessableEntityError |
| 429 | RateLimitError |
| >=500 | InternalServerError |
| N/A | APIConnectionError |
Certain errors will be automatically retried 2 times by default, with a short exponential backoff. Connection errors (for example, due to a network connectivity problem), 409 Conflict, 429 Rate Limit, and >=500 Internal errors will all be retried by default.
You can use the max_retries option to configure or disable this:
from ctxos import Ctxos
# Configure the default for all requests:
client = Ctxos(
# default is 2
max_retries=0,
)
# Or, configure per-request:
client.with_options(max_retries=5).complete.create(
prompt="Can you help me effectively ask for a raise at work?",
model="ctxos-1",
)Requests time out after 60 seconds by default. You can configure this with a timeout option,
which accepts a float or an httpx.Timeout:
import httpx
from ctxos import Ctxos
# Configure the default for all requests:
client = Ctxos(
# default is 60s
timeout=20.0,
)
# More granular control:
client = Ctxos(
timeout=httpx.Timeout(60.0, read=5.0, write=10.0, connect=2.0),
)
# Override per-request:
client.with_options(timeout=5 * 1000).complete.create(
prompt="Where can I get a good coffee in my neighbourhood?",
model="ctxos-1",
)On timeout, an APITimeoutError is thrown.
Note that requests which time out will be retried twice by default.
If you need to, you can override it by setting default headers per-request or on the client object.
Be aware that doing so may result in incorrect types and other unexpected or undefined behavior in the SDK.
from ctxos import Ctxos
ctxos = Ctxos(
default_headers={"ctxos-version": "My-Custom-Value"},
)You can configure the following keyword arguments when instantiating the client:
import httpx
from ctxos import Ctxos
ctxos = Ctxos(
# Use a custom base URL
base_url="http://my.test.server.example.com:8083",
proxies="http://my.test.proxy.example.com",
transport=httpx.HTTPTransport(local_address="0.0.0.0"),
)See the httpx documentation for information about the proxies and transport keyword arguments.
This package is in beta. Its internals and interfaces are not stable and subject to change without a major semver bump; please reach out if you rely on any undocumented behavior.
We are keen for your feedback; please open an issue with questions, bugs, or suggestions.
Python 3.9 or higher.