Python examples

Each example reads your key from the TAURIC_API_KEY environment variable, set as in the quickstart.

Choose a client

you want to	use	address
read data: quotes, history, filings, macro, news	HTTPX (any HTTP client)	`https://api.tauric.ai/v1/…`
run a full research report on a company	HTTPX, streamed	`/v1/research/agentic/analyze`
ask the managed copilot about the stored data	HTTPX	`/v1/agents/copilot/runs`
run the model with your own messages and tools	the OpenAI SDK	`/v1/chat/completions`, `/v1/responses`
find and restore your sessions of a Tauric application (session archive)	HTTPX	`/v1/applications/{application}/sessions`
give an agent the data as tools	the MCP SDK or your MCP client	`https://mcp.tauric.ai/mcp`

The OpenAI SDK covers only model inference; everything else is plain REST. There is no Tauric SDK.

pip install httpx openai "mcp>=2"

Read data

The REST examples on this page use this client and error helper, which turns the error envelope into an exception:

import os
import httpx

api = httpx.Client(
    base_url="https://api.tauric.ai",
    headers={"Authorization": f"Bearer {os.environ['TAURIC_API_KEY']}"},
    timeout=30.0,
)


def get(path, **params):
    response = api.get(path, params=params)
    if response.is_error:
        error = response.json()["error"]
        raise RuntimeError(f"{response.status_code} {error['code']}: {error['message']}")
    return response.json()

Data reads need the data entitlement (entitlements).

Find an instrument

Search by name or ticker to get an instrument's key (find an instrument).

for item in get("/v1/instruments", q="nvidia", limit=3)["data"]:
    print(item["key"], item["name"], item["maintenance"]["state"])

Get several quotes at once

POST /v1/quotes prices any mix of instruments in one request; an instrument key it cannot price is listed under errors (batch quotes).

keys = ["security:AAPL", "fx:EURUSD", "crypto:BTC-USDT", "series:us.treasury.yield_index.10y"]
response = api.post("/v1/quotes", json={"instruments": keys})
response.raise_for_status()
batch = response.json()
for key, quote in batch["quotes"].items():
    print(key, quote["price"], quote["instrument"]["currency"], quote["source_time"])
print("not priced:", batch["errors"])

Page back through daily history

A history read returns the newest window. coverage.continue_end is where the next older window ends, and it is null at the start of the history (price history).

page = get("/v1/securities/AAPL/history", period="1mo")
bars = page["bars"]
if page["coverage"]["continue_end"]:
    older = get("/v1/securities/AAPL/history", period="1mo", end=page["coverage"]["continue_end"])
    bars = older["bars"] + bars
print(len(bars), bars[0]["date"], bars[-1]["date"], bars[-1]["close"], bars[-1]["status"])

Read company financials

Values are reported per XBRL concept (us-gaap:…), newest period first, each citing its filing (filings and financials).

facts = get("/v1/securities/MSFT/financials", concepts="us-gaap:NetIncomeLoss,us-gaap:Assets",
            period_type="annual", limit=3)
for name, concept in facts["concepts"].items():
    for value in concept["values"]:
        print(name, value["period_end"], value["value"], concept["unit"], value["form"], value["accession"])

A concept with no value is listed under missing with its reason.

Read macro and rates series

Series are named by what they measure, and each observation carries its period (economic and market series). GET /v1/series lists every series.

cpi = get("/v1/series/us.cpi.all_items.sa/observations", limit=3)
print(cpi["series"]["name"], cpi["series"]["frequency"], "through", cpi["freshness"]["observed_through"])
for observation in cpi["observations"]:
    print(observation["period"]["label"], observation["value"])

Search and read the news

Search ranks articles by meaning, and each hit's id reads the full article (news and events).

hits = get("/v1/news/search", q="semiconductor export controls", limit=3)["data"]
for hit in hits:
    print(hit["article"]["article_time"], hit["article"]["title"])
article = get(f"/v1/news/{hits[0]['article']['id']}")
print(article["title"], article["url"])

Run a research report

Stream a research run on one company: the run event names it, unnamed data: frames carry the report text, and completion gives the verdict. It uses your AI pool.

import datetime
import json

payload = {"symbol": "MSFT", "date": datetime.date.today().isoformat()}
report, event = [], None
with api.stream("POST", "/v1/research/agentic/analyze", json=payload,
                timeout=httpx.Timeout(30.0, read=900.0)) as stream:
    for line in stream.iter_lines():
        if line.startswith("event: "):
            event = line[len("event: "):]
        elif line.startswith("data: ") and line != "data: [DONE]":
            data = json.loads(line[len("data: "):])
            if event == "run":
                run_id = data["id"]
            elif event == "completion":
                print("complete:", data["complete"], data["reason"], data["usage"]["charged_credits"])
            elif event is None and "choices" in data:
                report.append(data["choices"][0]["delta"].get("content") or "")
        elif line == "":
            event = None
print(len("".join(report)), "characters, run", run_id)

A completed report is stored: read it again with get(f"/v1/research/runs/{run_id}/report").

Ask the copilot

The copilot runs its own tools over the stored data, so you send only your message (copilot). The examples reuse the api client from read data, with a longer timeout.

run = api.post("/v1/agents/copilot/runs", json={"input": "What is the 10-year Treasury yield today?"},
               timeout=600).json()
print(run["status"], run["x_tauric"]["composition"])
for item in run["output"]:
    if item["type"] == "message":
        print(item["content"][0]["text"])

composition is explained in answer composition.

To print the answer as it is written, stream the run and read the text deltas:

import json

with api.stream("POST", "/v1/agents/copilot/runs", timeout=600,
                json={"input": "How did the 2-year yield move this week?", "stream": True}) as stream:
    for line in stream.iter_lines():
        if not line.startswith("data: "):
            continue
        event = json.loads(line[len("data: "):])
        if event["type"] == "response.output_text.delta":
            print(event["delta"], end="", flush=True)
        elif event["type"] in ("response.completed", "response.incomplete", "response.failed"):
            print("\n", event["response"]["status"])

Run the model with the OpenAI SDK

Set the OpenAI SDK's base_url to https://api.tauric.ai/v1; a tool call comes back for your code to run (model inference). This needs the ai entitlement and uses your AI credits.

import os
from openai import OpenAI

# max_retries=0: each retry would be a new charged request
ai = OpenAI(api_key=os.environ["TAURIC_API_KEY"], base_url="https://api.tauric.ai/v1", max_retries=0)
reply = ai.chat.completions.create(
    model="tauric",
    messages=[{"role": "system", "content": "Answer in one sentence."},
              {"role": "user", "content": "What does a yield-curve inversion usually signal?"}],
    max_tokens=200,
)
print(reply.choices[0].message.content)
print(reply.usage.completion_tokens, "output tokens")

ai.responses.create(...) works the same way. Model inference covers function tools for both.

Connect an agent over MCP

The data operations and research runs are also MCP tools (use with AI assistants).

import asyncio
import os
import httpx
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client


async def main():
    headers = {"Authorization": f"Bearer {os.environ['TAURIC_API_KEY']}"}
    async with httpx.AsyncClient(headers=headers, timeout=60) as http:
        async with streamable_http_client("https://mcp.tauric.ai/mcp", http_client=http) as (read, write):
            async with ClientSession(read, write) as session:
                await session.initialize()
                tools = await session.list_tools()
                print(len(tools.tools), "tools")
                result = await session.call_tool("get_security_quote", {"symbol": "AAPL"})
                if result.is_error:
                    print(result.content[0].text)
                else:
                    print(result.structured_content["price"], result.structured_content["source_time"])


asyncio.run(main())

An error result (is_error) carries the typed error as JSON text. A missing or malformed key is refused at the transport, and the SDK raises (tool errors).

Choose a client​

Read data​

Find an instrument​

Get several quotes at once​

Page back through daily history​

Read company financials​

Read macro and rates series​

Search and read the news​

Run a research report​

Ask the copilot​

Run the model with the OpenAI SDK​

Connect an agent over MCP​