HTTP va LLM API: Python'dan Claude bilan ishlash

uv add httpx

import httpx
 
resp = httpx.get(
    "https://api.github.com/search/repositories",
    params={"q": "language:python stars:>50000", "per_page": 3},   # URL'ga ?q=...&per_page=3
    headers={"Accept": "application/vnd.github+json"},
    timeout=10.0,                                                   # HAR DOIM timeout
)
resp.raise_for_status()          # 4xx/5xx bo'lsa — httpx.HTTPStatusError
for repo in resp.json()["items"]:
    print(repo["full_name"], repo["stargazers_count"])

with httpx.Client(base_url="https://api.example.com", timeout=10, headers={"Authorization": f"Bearer {token}"}) as client:
    users = client.get("/users").json()
    client.post("/events", json={"type": "login", "user_id": 42})    # json= — avtomatik serializatsiya

import os
 
import httpx
 
resp = httpx.post(
    "https://api.anthropic.com/v1/messages",
    headers={
        "x-api-key": os.environ["ANTHROPIC_API_KEY"],
        "anthropic-version": "2023-06-01",
        "content-type": "application/json",
    },
    json={
        "model": "claude-opus-5-5",
        "max_tokens": 1024,
        "messages": [{"role": "user", "content": "Python'da list va tuple farqi nima? Qisqa."}],
    },
    timeout=60,
)
resp.raise_for_status()
data = resp.json()
print(data["content"][0]["text"])
print(data["usage"])          # {'input_tokens': ..., 'output_tokens': ...}

uv add anthropic

# .env
ANTHROPIC_API_KEY=sk-ant-...

import anthropic
 
client = anthropic.Anthropic()        # kalitni ANTHROPIC_API_KEY'dan o'zi o'qiydi
 
message = client.messages.create(
    model="claude-opus-5-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Python'da list va tuple farqi nima? Qisqa."}],
)
 
for block in message.content:
    if block.type == "text":
        print(block.text)

message.id             # 'msg_01...'
message.model          # 'claude-opus-5-5'
message.stop_reason    # 'end_turn' | 'max_tokens' | 'tool_use' | 'refusal' | ...
message.usage          # input_tokens, output_tokens, cache_read_input_tokens, ...
message.content        # bloklar ro'yxati: text, thinking, tool_use...

SYSTEM = (
    "Sen tajribali Python o'qituvchisisan. O'zbek tilida, qisqa va aniq javob ber. "
    "Har javobda bitta kichik kod misoli bo'lsin."
)
 
history: list[dict] = []
 
 
def ask(question: str) -> str:
    history.append({"role": "user", "content": question})
    msg = client.messages.create(
        model="claude-opus-5-5",
        max_tokens=2048,
        system=SYSTEM,
        messages=history,
    )
    answer = "".join(b.text for b in msg.content if b.type == "text")
    history.append({"role": "assistant", "content": answer})
    return answer
 
 
ask("Dekorator nima?")
ask("Parametrli versiyasini ko'rsat")     # model oldingi savolni "biladi" — chunki tarix yuborildi

with client.messages.stream(
    model="claude-opus-5-5",
    max_tokens=4096,
    messages=[{"role": "user", "content": "Python'dagi GIL haqida tushuntir"}],
) as stream:
    for text in stream.text_stream:          # generator (7-bob!) — har bo'lak kelishi bilan
        print(text, end="", flush=True)
    final = stream.get_final_message()       # to'liq Message obyekti: usage, stop_reason
 
print()
print(final.usage.output_tokens, final.stop_reason)

def cost_usd(usage, price_in: float = 4.0, price_out: float = 20.0) -> float:
    return (usage.input_tokens * price_in + usage.output_tokens * price_out) / 1_000_000
 
cost_usd(message.usage)       # masalan, 0.0042

count = client.messages.count_tokens(
    model="claude-opus-5-5",
    messages=[{"role": "user", "content": long_document}],
)
count.input_tokens

response = client.messages.create(
    model="claude-opus-5-5",
    max_tokens=2048,
    system=[
        {"type": "text", "text": "Sen hujjat bo'yicha savollarga javob beruvchi yordamchisan."},
        {"type": "text", "text": big_document, "cache_control": {"type": "ephemeral"}},   # shu yergacha keshlanadi
    ],
    messages=[{"role": "user", "content": question}],
)
response.usage.cache_creation_input_tokens   # birinchi marta — keshga yozildi
response.usage.cache_read_input_tokens       # keyingi so'rovlarda — keshdan o'qildi

import anthropic
 
client = anthropic.Anthropic(max_retries=4, timeout=60.0)   # SDK 408/429/5xx va tarmoq xatolarida o'zi qayta urinadi
 
try:
    message = client.messages.create(model="claude-opus-5-5", max_tokens=1024, messages=msgs)
except anthropic.AuthenticationError:
    raise SystemExit("API kaliti noto'g'ri — ANTHROPIC_API_KEY'ni tekshiring")
except anthropic.BadRequestError as e:
    log.error("So'rov noto'g'ri: %s", e.message)        # qayta urinish foydasiz — kodni tuzating
    raise
except anthropic.RateLimitError:
    log.warning("Limit — SDK retry'lari ham yordam bermadi")
    raise
except anthropic.APIStatusError as e:
    log.error("API xatosi %s: %s", e.status_code, e.message)
    raise
except anthropic.APIConnectionError:
    log.error("Tarmoq xatosi")
    raise

from typing import Literal
 
import anthropic
from pydantic import BaseModel, Field
 
 
class Review(BaseModel):
    sentiment: Literal["positive", "negative", "neutral"]
    topics: list[str] = Field(description="Sharhda tilga olingan mavzular, masalan: yetkazib berish, narx")
    summary: str = Field(description="Bir gaplik xulosa, o'zbek tilida")
 
 
client = anthropic.Anthropic()
 
response = client.messages.parse(
    model="claude-opus-5-5",
    max_tokens=1024,
    messages=[{
        "role": "user",
        "content": "Sharhni tahlil qil: 'Telefon zo'r, lekin yetkazib berish 2 hafta davom etdi.'",
    }],
    output_format=Review,
)
 
review = response.parsed_output       # Review obyekti — tiplar tekshirilgan
review.sentiment                      # masalan, 'neutral'
review.topics                         # masalan, ['telefon sifati', 'yetkazib berish']

import json
 
import anthropic
 
client = anthropic.Anthropic()
 
TOOLS = [
    {
        "name": "get_weather",
        "description": "Berilgan shahar uchun joriy ob-havoni qaytaradi.",
        "input_schema": {
            "type": "object",
            "properties": {"city": {"type": "string", "description": "Shahar nomi, masalan: Toshkent"}},
            "required": ["city"],
        },
    },
    {
        "name": "convert_currency",
        "description": "Summani bir valyutadan boshqasiga o'tkazadi.",
        "input_schema": {
            "type": "object",
            "properties": {
                "amount": {"type": "number"},
                "from_currency": {"type": "string", "enum": ["UZS", "USD", "EUR"]},
                "to_currency": {"type": "string", "enum": ["UZS", "USD", "EUR"]},
            },
            "required": ["amount", "from_currency", "to_currency"],
        },
    },
]
 
RATES = {"USD": 12_650.0, "EUR": 13_900.0, "UZS": 1.0}       # soxta kurslar
 
 
def get_weather(city: str) -> dict:
    return {"city": city, "temp_c": 21, "condition": "quyoshli"}      # haqiqiy loyihada — API chaqiruvi
 
 
def convert_currency(amount: float, from_currency: str, to_currency: str) -> dict:
    result = amount * RATES[from_currency] / RATES[to_currency]
    return {"result": round(result, 2), "currency": to_currency}
 
 
HANDLERS = {"get_weather": get_weather, "convert_currency": convert_currency}
 
 
def run_tool(name: str, args: dict) -> tuple[str, bool]:
    """Asbobni bajaradi. (natija_json, xatomi) qaytaradi."""
    try:
        return json.dumps(HANDLERS[name](**args), ensure_ascii=False), False
    except Exception as e:                       # xatoni modelga qaytaramiz — u tuzatishga urinadi
        return f"Xato: {e!r}", True
 
 
def agent(question: str, max_steps: int = 10) -> str:
    messages = [{"role": "user", "content": question}]
    for _ in range(max_steps):                   # cheksiz sikldan himoya
        response = client.messages.create(
            model="claude-opus-5-5", max_tokens=4096, tools=TOOLS, messages=messages,
        )
        messages.append({"role": "assistant", "content": response.content})
 
        if response.stop_reason != "tool_use":
            return "".join(b.text for b in response.content if b.type == "text")
 
        results = []
        for block in response.content:
            if block.type == "tool_use":
                output, is_error = run_tool(block.name, block.input)
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,         # qaysi chaqiruvga javob ekanini bog'laydi
                    "content": output,
                    "is_error": is_error,
                })
        messages.append({"role": "user", "content": results})    # BARCHA natijalar bitta xabarda
    raise RuntimeError("Agent qadamlar limitiga yetdi")
 
 
print(agent("Samarqandda havo qanday? Va 250 dollar necha so'm bo'ladi?"))

from anthropic import beta_tool
 
 
@beta_tool
def get_weather(city: str) -> str:
    """Berilgan shahar uchun joriy ob-havoni qaytaradi.
 
    Args:
        city: Shahar nomi, masalan: Toshkent.
    """
    return json.dumps({"city": city, "temp_c": 21, "condition": "quyoshli"}, ensure_ascii=False)
 
 
runner = client.beta.messages.tool_runner(
    model="claude-opus-5-5",
    max_tokens=4096,
    tools=[get_weather],
    messages=[{"role": "user", "content": "Buxoroda havo qanday?"}],
)
for message in runner:             # har qadamda Message; asboblar avtomatik bajariladi
    for block in message.content:
        if block.type == "text":
            print(block.text)

import base64
from pathlib import Path
 
image_b64 = base64.standard_b64encode(Path("chek.jpg").read_bytes()).decode()
 
message = client.messages.create(
    model="claude-opus-5-5",
    max_tokens=1024,
    messages=[{
        "role": "user",
        "content": [
            {"type": "image", "source": {"type": "base64", "media_type": "image/jpeg", "data": image_b64}},
            {"type": "text", "text": "Chekdagi jami summa va sanani ayt."},
        ],
    }],
)

import asyncio
 
from anthropic import AsyncAnthropic
 
 
async def summarize_all(texts: list[str], concurrency: int = 8) -> list[str]:
    sem = asyncio.Semaphore(concurrency)
    async with AsyncAnthropic() as client:
 
        async def one(text: str) -> str:
            async with sem:
                msg = await client.messages.create(
                    model="claude-opus-5-5",
                    max_tokens=300,
                    output_config={"effort": "low"},      # oddiy vazifa — kam "o'ylash", tezroq va arzonroq
                    messages=[{"role": "user", "content": f"Bir gapda xulosa qil:\n\n{text}"}],
                )
            return "".join(b.text for b in msg.content if b.type == "text")
 
        return await asyncio.gather(*(one(t) for t in texts))

"""chat.py — terminalda Claude bilan streaming suhbat."""
 
import os
import sys
 
import anthropic
 
MODEL = os.environ.get("MODEL", "claude-opus-5-5")
SYSTEM = "Sen Python bo'yicha yordamchisan. O'zbek tilida, aniq va qisqa javob ber."
MAX_TURNS = 20
PRICE_IN, PRICE_OUT = 4.0, 20.0
 
 
def main() -> None:
    client = anthropic.Anthropic(max_retries=3)
    history: list[dict] = []
    total_cost = 0.0
    print(f"{MODEL} bilan suhbat. Chiqish: Ctrl+D yoki 'chiq'.\n")
 
    while True:
        try:
            user = input("Siz: ").strip()
        except (EOFError, KeyboardInterrupt):
            break
        if not user:
            continue
        if user.lower() in {"chiq", "exit"}:
            break
 
        history.append({"role": "user", "content": user})
        print("Claude: ", end="", flush=True)
        try:
            with client.messages.stream(
                model=MODEL, max_tokens=4096, system=SYSTEM, messages=history[-2 * MAX_TURNS:],
            ) as stream:
                for text in stream.text_stream:
                    print(text, end="", flush=True)
                final = stream.get_final_message()
        except anthropic.APIError as e:
            history.pop()                                  # muvaffaqiyatsiz savolni tarixdan olib tashlaymiz
            print(f"\n[xato: {type(e).__name__}] qayta urinib ko'ring.", file=sys.stderr)
            continue
 
        answer = "".join(b.text for b in final.content if b.type == "text")
        if final.stop_reason == "max_tokens":
            print("\n[javob uzunlik limitida kesildi]", end="")
        history.append({"role": "assistant", "content": answer})
 
        u = final.usage
        cost = (u.input_tokens * PRICE_IN + u.output_tokens * PRICE_OUT) / 1e6
        total_cost += cost
        print(f"\n  [{u.input_tokens}→{u.output_tokens} token, ${cost:.4f}]\n")
 
    print(f"\nJami: ${total_cost:.4f}")
 
 
if __name__ == "__main__":
    main()