openai-iztro-agents

Build your own Ziwei (Purple Star Astrology / 紫微斗数) or Qimen agent in Python.

pip install openai-iztro-agents → import iztro_agents

🟨 Working in JavaScript / TypeScript? See the sibling package openai-iztro-agents-js — same design, JS conventions.

A thin layer on top of the OpenAI Agents SDK:

• The hosted Ziwei and Qimen models and their astrology tools run on the server (hidden) — exposed as stock SDK models.
• Your own function tools, MCP servers, and human-in-the-loop run locally via the standard Runner.
• Conversation memory lives on the server via ChatSession (the OpenAI Conversations–style session).

You write ordinary OpenAI Agents SDK code — Agent, Runner, @function_tool, agents.mcp, tool_choice, needs_approval — and point the model at Ziwei or Qimen.

Install

pip install openai-iztro-agents

Get an API key (sk_ziwei_*) from the developer console.

Set it once in the server environment so the Agent and ChatSession factories can read it:

# macOS / Linux
export ZIWEI_API_KEY="sk_ziwei_..."

# PowerShell
$env:ZIWEI_API_KEY="sk_ziwei_..."

Quickstart

import asyncio
from agents import Runner
from iztro_agents import iztro_ziwei_agent, ChatSession, function_tool

@function_tool
def add_to_calendar(date: str, title: str) -> str:
    """Add an event to the user's calendar. Runs locally."""
    return f"Added '{title}' on {date}"

async def main():
    agent = iztro_ziwei_agent(tools=[add_to_calendar], api_key="sk_ziwei_...")
    session = ChatSession(external_user_id="user_42")
    result = await Runner.run(
        agent,
        "I was born 1990-06-15 at 10am, male. Pick a good day next week and add it to my calendar.",
        session=session,
    )
    print(result.final_output)

asyncio.run(main())

iztro_ziwei_agent(...) returns a stock agents.Agent whose model is the hosted Ziwei agent — so everything from the OpenAI Agents SDK works unchanged (result.new_items, Runner.run_streamed, handoffs, tracing, …).

Model settings

Both hosted models use the same supported subset of the native OpenAI Agents SDK ModelSettings. Deep reasoning and sampling are separate modes:

from agents import ModelSettings
from openai.types.shared import Reasoning
from iztro_agents import iztro_qimen_agent, iztro_ziwei_agent

deep_settings = ModelSettings(
    reasoning=Reasoning(effort="high"),
    max_tokens=384000,
    tool_choice="auto",
    parallel_tool_calls=True,
    metadata={"current_datetime": "2026-07-20T14:30:00+08:00"},
    extra_body={"language": "vi"},
)

fast_settings = ModelSettings(
    reasoning=Reasoning(effort="none"),
    temperature=0.4,  # Or top_p=0.9; do not set both.
    extra_body={"language": "vi"},
)

ziwei = iztro_ziwei_agent(model_settings=deep_settings)
qimen = iztro_qimen_agent(model_settings=fast_settings)

Omit reasoning, or use none, minimal, or low, for the faster non-thinking path. medium, high, and xhigh use the same high deep-reasoning path. temperature and top_p work only on the non-thinking path and are mutually exclusive. DeepSeek does not support frequency_penalty or presence_penalty; the hosted API rejects them instead of silently ignoring them. Omit max_tokens to keep the current 384,000-token default output capacity.

Non-thinking mode prioritizes speed and can miss or mis-associate details in a complex chart. Use reasoning=Reasoning(effort="high") for cross-palace Ziwei synthesis, multiple fortune layers, or Qimen decisions that combine chart and timing evidence. Deep reasoning is generally more reliable for these readings, but it does not guarantee correctness.

Supported response languages are zh, en, ko, ja, and vi; setting extra_body.language keeps all user-facing prose and translated terminology in that language without mixing. For live Qimen requests, create metadata.current_datetime from the user's local clock on each turn rather than keeping the fixed example value.

See the complete Model settings support matrix, including raw HTTP names and upstream SDK fields that are not hosted-model controls.

Qimen model

iztro-qimen-v3 is a hosted Qimen Dunjia model for a time-sensitive decision about one concrete matter. Use it for questions such as:

• Should we advance this partnership now, negotiate first, or pause?
• How is this interview, offer, launch, trip, or relationship decision likely to develop?
• If the matter can move forward, which dates are the meaningful action windows?

It casts the chart from the question time, so it does not need a birth date, birth hour, or gender. Use iztro-ziwei-v3 instead for natal personality, compatibility, or long-range life and fortune analysis.

Model	Best for	Required input
iztro-qimen-v3	One current event, decision, outcome, and optional timing	The concrete situation and question time
iztro-ziwei-v3	Natal profile, compatibility, and longer-term fortune cycles	Birth date, birth time, and gender

Qimen request rules

1. Ask about one concrete matter and put unrelated decisions in separate runs.
2. Give the current facts, choices, and constraint, then ask one explicit decision.
3. When timing matters, ask for an action window and pass the user's local question time.

Timing results are candidate trigger windows to interpret with the complete answer, not guaranteed outcomes.

Qimen quickstart

Use iztro_qimen_agent(...) for a ready-to-run stock Agent, or iztro_qimen_model(...) when constructing the Agent yourself:

import asyncio
import os

from agents import ModelSettings, Runner
from openai.types.shared import Reasoning
from iztro_agents import iztro_qimen_agent

async def main() -> None:
    agent = iztro_qimen_agent(
        api_key=os.environ["ZIWEI_API_KEY"],
        # Optional: pin the user's local question time for reproducible charts.
        # If omitted, the service uses the request time.
        model_settings=ModelSettings(
            reasoning=Reasoning(effort="high"),
            metadata={"current_datetime": "2026-07-20T14:30:00+08:00"}
        ),
    )
    result = await Runner.run(
        agent,
        (
            "我们正在谈一项渠道合作，已经沟通两次，但分成和上线时间还没定。"
            "现在适合主动推进、继续谈判，还是暂缓？如果适合推进，请给出近期时间窗口和行动建议。"
        ),
    )
    print(result.final_output)

asyncio.run(main())

For a strong request, describe the current situation, ask one decision, and say whether you need timing. Put unrelated matters in separate runs so each receives its own chart. See the complete 12_qimen_decision.py example and compare both public models in the Models guide.

Public Iztro calculation names are available through Iztro tool events. Your own function tools, MCP servers, and human-in-the-loop continue to use the normal OpenAI Agents SDK interfaces.

Iztro calculation events

The wrapper exposes the public calculation names returned by the API as Iztro tool events:

result = await Runner.run(agent, "用奇门起局并判断应期。")
event = result.raw_responses[-1].tool_event
print(event.type, event.tools)  # tool_event ['qimen-qigua', 'qimen-yingqi']

The complete list of public return values is in the Models guide. Do not depend on undocumented names or infer internal implementation from these values.

Streaming can include IztroToolEvent. The older .iztro_tools, last_iztro_tools, and IztroToolsStreamEvent names still work for compatibility, but new code should use tool_event / IztroToolEvent.

Conversation memory & resume (ChatSession)

History is stored on the server with a server-generated id, owned by your external_user_id:

from iztro_agents import ChatSession, list_user_conversations

session = ChatSession(external_user_id="user_42")     # ZIWEI_API_KEY from env
await Runner.run(agent, "My name is Alice.", session=session)
await Runner.run(agent, "What's my name?", session=session)   # remembers

conv_id = session.session_id          # save to resume later
ChatSession(conversation_id=conv_id)  # resume

# Manage a user's chats:
await list_user_conversations("user_42")

session_id precedence: explicit conversation_id > a server-assigned id created lazily on first use.

Fork a complete conversation, or copy only the first N SDK session items before continuing with replacement text:

forked = await session.fork()                 # copy the whole conversation
edited = await session.fork(item_count=4)    # copy items 0..3, then branch
await Runner.run(agent, "Use this edited question instead", session=edited)

The runnable ChatSession full-stack demo combines this with conversation lists, titles, deletion, history editing, live tool/chart indicators, and SSE streaming while keeping the API key on the backend.

Tool-call modes

Your tools use the SDK's native controls; the iztro tools are hidden (toggle with enable_iztro_call):

from agents import ModelSettings
agent = iztro_ziwei_agent(
    tools=[...],
    model_settings=ModelSettings(tool_choice="auto", parallel_tool_calls=True),
)

Human-in-the-loop (native SDK)

@function_tool(needs_approval=True)
def send_email(to: str, subject: str, body: str) -> str: ...

result = await Runner.run(agent, "...")
while result.interruptions:          # SDK pauses before the tool runs
    state = result.to_state()
    for item in result.interruptions:
        state.approve(item)          # or state.reject(item)
    result = await Runner.run(agent, state)

MCP servers

from agents.mcp import MCPServerStdio
weather = MCPServerStdio(params={"command": "uvx", "args": ["mcp-server-weather"]})
agent = iztro_ziwei_agent(mcp_servers=[weather], api_key=KEY)

Testing

# Fast, deterministic, offline (no key) — mocks the model + conversation HTTP:
pytest            # runs the whole offline suite (the live test self-skips)

# Live end-to-end against a deployed backend (opt-in):
ZIWEI_API_KEY=sk_ziwei_... pytest tests/test_live.py -v -s
# defaults to dev; prod via ZIWEI_BASE_URL=https://chat-api.iztro.com

The offline suite covers a wide range of scenarios — each test file is written so a scenario can graduate into an examples/ script:

File	What it exercises
tests/test_tool_loops.py	plain chat, single/parallel/sequential tool calls, typed args, local-tool errors, unicode, tool_choice/parallel_tool_calls, and that iztro tools stay hidden
tests/test_human_in_the_loop.py	native needs_approval flow — approve, reject, and a mixed approve+reject turn
tests/test_streaming.py	Runner.run_streamed token deltas reassembling into final_output
tests/test_session.py	ChatSession memory — lazy id, add/get/pop/clear, multi-turn, ownership + listing, resume
tests/test_factories.py	credential/base-url resolution, /v2 suffix, and SDK arg passthrough

Shared offline backends live in tests/_mock.py (a fake chat-completions endpoint and an in-memory conversation store).

Notes

• Birth details are gathered by the Ziwei agent through the conversation — there is no birth_info parameter.
• The backend currently streams an answer as a single chunk (not token-by-token); Runner.run_streamed works but token-level streaming is a future backend enhancement.
• Streaming together with developer tools is not yet supported — use non-streaming Runner.run for tool loops.
• Multi-turn tool loops re-send the prompt each round, so they cost more tokens.