DEV Community: Serhii Kalyna

Streaming Responses with Claude API in Python (2026)

Serhii Kalyna — Fri, 12 Jun 2026 13:09:50 +0000

Originally published at kalyna.pro

Streaming sends Claude's response token by token as it's generated, instead of waiting for the full completion before showing anything. For a chat UI this is the difference between a user staring at a spinner for several seconds and seeing the first words appear within a few hundred milliseconds. The Claude API Tutorial introduces the basic stream.text_stream helper — this guide covers the full picture: the raw event stream, async streaming, error handling, and a complete FastAPI endpoint that streams Claude's output to a browser.

Prerequisites

pip install anthropic
# for the API endpoint example later:
pip install fastapi uvicorn

The Simple Way: text_stream

from anthropic import Anthropic

client = Anthropic()

with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Write a haiku about debugging."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

    final_message = stream.get_final_message()

print(f"\n\nstop_reason: {final_message.stop_reason}")
print(f"output tokens: {final_message.usage.output_tokens}")

stream.get_final_message() returns the same Message object you'd get from a non-streaming call — complete content, stop_reason, and usage — without manually reassembling it from chunks.

The Raw Event Stream

with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Write a haiku about debugging."}],
) as stream:
    for event in stream:
        print(event.type)

Event types, in order:

message_start — initial Message shell with usage.input_tokens
content_block_start — a new content block begins (text, tool_use, etc.)
content_block_delta — incremental content: text_delta (.text), input_json_delta (.partial_json, for tool inputs), or thinking_delta
content_block_stop — the block is complete
message_delta — stop_reason and updated usage.output_tokens
message_stop — stream finished

with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Write a haiku about debugging."}],
) as stream:
    for event in stream:
        if event.type == "content_block_delta" and event.delta.type == "text_delta":
            print(event.delta.text, end="", flush=True)
        elif event.type == "message_delta":
            print(f"\n[tokens so far: {event.usage.output_tokens}]", end="")

Async Streaming

import asyncio
from anthropic import AsyncAnthropic

client = AsyncAnthropic()


async def main():
    async with client.messages.stream(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[{"role": "user", "content": "Write a haiku about debugging."}],
    ) as stream:
        async for text in stream.text_stream:
            print(text, end="", flush=True)


asyncio.run(main())

Building a Streaming API Endpoint (FastAPI + SSE)

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from anthropic import AsyncAnthropic

app = FastAPI()
client = AsyncAnthropic()


@app.get("/chat")
async def chat(message: str):
    async def event_stream():
        async with client.messages.stream(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            messages=[{"role": "user", "content": message}],
        ) as stream:
            async for text in stream.text_stream:
                yield f"data: {text}\n\n"

        yield "event: done\ndata: {}\n\n"

    return StreamingResponse(
        event_stream(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )

X-Accel-Buffering: no stops nginx from buffering the whole response — without it, "streaming" arrives in one burst at the end. On the frontend, read with fetch + a ReadableStream reader, or EventSource for GET endpoints.

Handling Errors and Interruptions

import anthropic

try:
    with client.messages.stream(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[{"role": "user", "content": "Write a haiku about debugging."}],
    ) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
except anthropic.APIConnectionError:
    print("\n[connection lost — showing partial response]")
except anthropic.RateLimitError:
    print("\n[rate limited — retry shortly]")
except anthropic.APIStatusError as e:
    print(f"\n[API error {e.status_code}]")

If the client disconnects mid-response, exit the generator early so the SDK closes the stream — this stops billing for output tokens generated into the void. For long generations, check await request.is_disconnected() periodically and break if true.

Streaming with Tool Use

Text still arrives via text_delta, tool arguments arrive incrementally via input_json_delta, and stream.get_final_message() gives fully-parsed tool_use blocks once the stream ends. See Claude API Function Calling for the complete tool-use loop — it works unchanged whether calls are streamed or not.

Best Practices

Use get_final_message() for stop_reason/usage instead of accumulating message_delta manually
Use AsyncAnthropic in web backends — a sync stream blocks the event loop
Set Cache-Control: no-cache and X-Accel-Buffering: no for SSE behind a proxy
Detect client disconnects and stop generation early
Streaming doesn't change pricing — tokens are billed the same either way
Handle APIConnectionError, RateLimitError, and APIStatusError explicitly

Summary

stream.text_stream yields plain text chunks for display
Raw events: message_start, content_block_start, content_block_delta, content_block_stop, message_delta, message_stop
get_final_message() returns the complete Message after streaming
AsyncAnthropic + async with/async for for non-blocking backends
FastAPI StreamingResponse + async generator → SSE to the browser
Tool use streams the same way; input_json_delta carries tool arguments

Claude Prompt Caching: How to Cut API Costs (2026)

Serhii Kalyna — Fri, 12 Jun 2026 13:06:29 +0000

Originally published at kalyna.pro

If your app sends the same large system prompt, tool definitions, or document context on every request, you're paying full price to re-process those tokens every single time. Prompt caching lets Claude reuse the processed representation of a prompt prefix across requests — cache hits cost 90% less than normal input tokens. This guide covers how caching actually works, the pricing math for writes vs reads, where to place cache breakpoints, and worked cost examples for RAG apps and agents.

Prerequisites

pip install anthropic

See the Claude API Tutorial if you're new to the Messages API, and Claude API Pricing Explained for base per-model pricing.

How Prompt Caching Works

Caching works on prefixes. A request is processed top to bottom — system prompt, then tool definitions, then messages. You mark a cache breakpoint by adding "cache_control": {"type": "ephemeral"} to a content block. Everything from the start of the prompt up to and including that block is cached as a unit.

On the next request, if the prefix up to a breakpoint is byte-for-byte identical to a cached prefix, Claude reads the cached version instead of reprocessing it — you only pay full price for whatever comes after the last matching breakpoint. If anything before a breakpoint changes, that cache entry misses and gets rewritten.

Cached prefixes expire after 5 minutes by default, refreshed on every cache hit
An extended 1-hour TTL is available for less frequent traffic
Each cacheable block must be at least 1024 tokens for Sonnet and Opus, or 2048 tokens for Haiku — shorter blocks aren't cached
Up to 4 cache breakpoints per request

Pricing: Cache Writes vs Cache Reads

Using Claude Sonnet 4.6 ($3 / $15 per 1M tokens, input/output) as an example:

Base input tokens — $3.00 / 1M
Cache write, 5-minute TTL (default) — 1.25× base: $3.75 / 1M
Cache write, 1-hour TTL (extended) — 2× base: $6.00 / 1M
Cache read (cache hit) — 0.1× base, a 90% discount: $0.30 / 1M
Output tokens — unaffected: $15.00 / 1M as usual

The same multipliers apply to every model's base price — for Haiku 4.5 ($0.25 / $1.25 per 1M) a cache read costs $0.025 / 1M; for Opus 4.7 ($15 / $75 per 1M) it costs $1.50 / 1M.

Cost Example: A RAG Chatbot

A RAG chatbot sends a 10,000-token retrieved-context block as the cached prefix, handling 100 requests/day at a steady pace.

Without caching: 100 × 10,000 × $3.00/1M = $3.00/day

With caching: first request writes the cache (10,000 × $3.75/1M = $0.0375). The remaining 99 read from cache (99 × 10,000 × $0.30/1M = $0.297). Total: ≈ $0.33/day — about an 89% reduction, compounding with every extra request.

Basic Example: Caching a System Prompt

from anthropic import Anthropic

client = Anthropic()

LARGE_SYSTEM_PROMPT = open("knowledge_base.md").read()  # 8,000+ tokens

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": LARGE_SYSTEM_PROMPT,
            "cache_control": {"type": "ephemeral"},
        }
    ],
    messages=[{"role": "user", "content": "Summarize the refund policy."}],
)

print(response.usage)
# Usage(input_tokens=12, cache_creation_input_tokens=8000, cache_read_input_tokens=0, ...)

The first call writes the cache. Send the same system prompt again within 5 minutes, and cache_read_input_tokens=8000, cache_creation_input_tokens=0 — those tokens now cost 90% less.

Caching Tool Definitions

Tool schemas count as input tokens on every call, including every step of a tool-use loop. Cache them by adding cache_control to the last tool in the list:

tools = [
    {"name": "get_stock_price", "description": "...", "input_schema": {...}},
    {"name": "get_exchange_rate", "description": "...", "input_schema": {...}},
    {
        "name": "get_current_time",
        "description": "Get the current date and time in UTC.",
        "input_schema": {"type": "object", "properties": {}},
        "cache_control": {"type": "ephemeral"},
    },
]

This caches all tool definitions as one prefix block — high leverage for agents, since tools is resent unchanged on every loop step.

Caching Long Documents

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[
        {
            "role": "user",
            "content": [
                {
                    "type": "text",
                    "text": f"<document>{long_document}</document>",
                    "cache_control": {"type": "ephemeral"},
                },
                {
                    "type": "text",
                    "text": "What are the payment terms in section 4?",
                },
            ],
        }
    ],
)

Follow-up questions about the same document, sent within 5 minutes, only pay full price for the new question.

Extended TTL: 1-Hour Cache

{
    "type": "text",
    "text": LARGE_SYSTEM_PROMPT,
    "cache_control": {"type": "ephemeral", "ttl": "1h"},
}

The 1-hour cache costs 2× base to write (vs 1.25× for 5-minute), reads are still 0.1×. It pays off once a cached prefix is reused at least twice within the hour. For traffic denser than every 5 minutes, the default TTL is usually cheaper.

Multiple Cache Breakpoints

Up to 4 breakpoints per request:

system=[
    {
        "type": "text",
        "text": STATIC_INSTRUCTIONS,           # rarely changes — shared across all users
        "cache_control": {"type": "ephemeral"},
    },
    {
        "type": "text",
        "text": USER_PROFILE_CONTEXT,          # changes per user
        "cache_control": {"type": "ephemeral"},
    },
]

Claude checks for the longest matching cached prefix. Ordering from least-to-most volatile means a change to USER_PROFILE_CONTEXT doesn't invalidate STATIC_INSTRUCTIONS.

Best Practices

Put stable content first, volatile content last — only what comes after the last cache hit is billed at full price
Don't cache below ~1024 tokens (2048 for Haiku) — no discount applies
Cache growing conversation history for multi-turn chats
Cache the tools array for agents
Track usage.cache_read_input_tokens vs cache_creation_input_tokens — aim for 90%+ hit rate on steady traffic
Use 1-hour TTL only when requests are >5 min apart but reuse the prefix within an hour

Summary

Prompt caching reuses a processed prompt prefix, marked via cache_control: {"type": "ephemeral"}
Cache reads cost 0.1× base price; 5-min writes cost 1.25×, 1-hour writes cost 2×
Minimum cacheable size: 1024 tokens (2048 for Haiku)
Default TTL is 5 minutes (refreshed on hits); use "ttl": "1h" for sparser traffic
Up to 4 breakpoints per request, ordered least-to-most volatile
Best targets: system prompts, agent tool definitions, RAG documents, conversation history

Claude API Function Calling: Complete Guide to Tool Use (2026)

Serhii Kalyna — Fri, 12 Jun 2026 11:41:13 +0000

Originally published at kalyna.pro

Function calling — what Anthropic calls tool use — lets Claude call code you write: query a database, hit an internal API, run a calculation, or check today's date. Claude never executes anything itself. It returns a structured request to call a specific tool with specific arguments, your code runs that tool, and you send the result back so Claude can continue. The Claude API Tutorial covers a single-tool example — this guide goes further: multi-step tool loops, parallel tool calls, forcing a specific tool, streaming tool inputs, and error handling, finishing with a complete multi-tool agent you can extend.

Prerequisites

pip install anthropic
export ANTHROPIC_API_KEY="sk-ant-..."

Defining a Tool

A tool definition has three parts: a name, a description, and an input_schema written as JSON Schema. Claude relies entirely on these three fields to decide whether to call the tool and how to fill in its arguments.

tools = [
    {
        "name": "get_stock_price",
        "description": "Get the current price of a stock by its ticker symbol.",
        "input_schema": {
            "type": "object",
            "properties": {
                "ticker": {
                    "type": "string",
                    "description": "Stock ticker symbol, e.g. 'AAPL' or 'GOOGL'",
                },
                "currency": {
                    "type": "string",
                    "enum": ["USD", "EUR", "GBP"],
                    "description": "Currency to return the price in. Defaults to USD.",
                },
            },
            "required": ["ticker"],
        },
    }
]

If Claude decides the tool is needed, the response has stop_reason == "tool_use" and response.content includes a tool_use block with name, input (already parsed as a dict), and a unique id.

The Tool-Use Loop

Tool use is a loop:

Call messages.create() with tools and the conversation so far
If stop_reason != "tool_use", Claude is done — return its text
Otherwise, execute every tool_use block in response.content and collect the results
Append Claude's response and a new user message containing tool_result blocks, then repeat

from anthropic import Anthropic

client = Anthropic()


def get_stock_price(ticker: str, currency: str = "USD") -> str:
    prices = {"AAPL": 230.15, "GOOGL": 178.32, "MSFT": 415.50}
    price = prices.get(ticker.upper())
    if price is None:
        return f"No data for ticker '{ticker}'"
    return f"{ticker.upper()}: {price} {currency}"


tools = [...]  # tool definition from above

messages = [{"role": "user", "content": "What's Apple's stock price?"}]

while True:
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        tools=tools,
        messages=messages,
    )

    if response.stop_reason != "tool_use":
        print(response.content[0].text)
        break

    messages.append({"role": "assistant", "content": response.content})

    tool_results = []
    for block in response.content:
        if block.type == "tool_use" and block.name == "get_stock_price":
            result = get_stock_price(**block.input)
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": result,
            })

    messages.append({"role": "user", "content": tool_results})

Parallel Tool Calls

Claude can request several tools in a single response — for example, a stock price and an exchange rate together. In that case response.content contains more than one tool_use block, each with its own id.

def get_exchange_rate(from_currency: str, to_currency: str) -> str:
    rates = {("USD", "EUR"): 0.92, ("EUR", "USD"): 1.09}
    rate = rates.get((from_currency.upper(), to_currency.upper()))
    if rate is None:
        return f"No rate for {from_currency} -> {to_currency}"
    return f"1 {from_currency.upper()} = {rate} {to_currency.upper()}"


tools.append({
    "name": "get_exchange_rate",
    "description": "Get the exchange rate between two currencies.",
    "input_schema": {
        "type": "object",
        "properties": {
            "from_currency": {"type": "string", "description": "3-letter code, e.g. 'USD'"},
            "to_currency": {"type": "string", "description": "3-letter code, e.g. 'EUR'"},
        },
        "required": ["from_currency", "to_currency"],
    },
})

The loop from the previous section already handles parallel calls — it iterates over every block and replies with one tool_result per tool_use_id. Order doesn't matter; Claude matches results to calls by tool_use_id.

Controlling Tool Choice

{"type": "auto"} — default. Claude decides whether a tool is needed.
{"type": "any"} — Claude must call one of the provided tools.
{"type": "tool", "name": "..."} — Claude must call this specific tool.
{"type": "none"} — Claude must not call any tool.

Forcing a specific tool is a clean way to get structured JSON output:

save_contact_tool = {
    "name": "save_contact",
    "description": "Save extracted contact information.",
    "input_schema": {
        "type": "object",
        "properties": {
            "name": {"type": "string", "description": "Full name of the contact"},
            "email": {"type": "string", "description": "Email address"},
            "phone": {"type": "string", "description": "Phone number, with country code if present"},
        },
        "required": ["name", "email"],
    },
}

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    tools=[save_contact_tool],
    tool_choice={"type": "tool", "name": "save_contact"},
    messages=[{
        "role": "user",
        "content": "Reach out to John Doe at john@example.com, phone +1 555-1234.",
    }],
)

contact = response.content[0].input
print(contact)
# {'name': 'John Doe', 'email': 'john@example.com', 'phone': '+1 555-1234'}

You can also pass "disable_parallel_tool_use": True inside tool_choice to guarantee exactly one tool call back.

Streaming Tool Inputs

with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    tools=tools,
    messages=messages,
) as stream:
    for event in stream:
        if event.type == "content_block_delta" and event.delta.type == "input_json_delta":
            print(event.delta.partial_json, end="", flush=True)

    final = stream.get_final_message()

for block in final.content:
    if block.type == "tool_use":
        print(block.name, block.input)

get_final_message() gives you fully-parsed tool_use blocks once the stream ends — you rarely need to assemble partial_json fragments yourself.

Error Handling in Tool Results

try:
    result = get_stock_price(**block.input)
    tool_results.append({
        "type": "tool_result",
        "tool_use_id": block.id,
        "content": result,
    })
except Exception as e:
    tool_results.append({
        "type": "tool_result",
        "tool_use_id": block.id,
        "content": f"Error: {e}",
        "is_error": True,
    })

Claude sees the error and adapts — retries with different arguments, tries another tool, or explains the failure.

Best Practices for Tool Definitions

Write descriptions like API docs for a new teammate
Use enum to constrain free-text values
Keep required minimal
One tool = one job — avoid mega-tools
Return concise, structured results — large blobs cost input tokens on every following call
Name tools with verbs (get_, search_, create_)

Complete Example: A Multi-Tool Agent

from anthropic import Anthropic
from datetime import datetime, timezone

client = Anthropic()

def get_stock_price(ticker: str, currency: str = "USD") -> str:
    prices = {"AAPL": 230.15, "GOOGL": 178.32, "MSFT": 415.50}
    price = prices.get(ticker.upper())
    if price is None:
        return f"No data for ticker '{ticker}'"
    return f"{ticker.upper()}: {price} {currency}"


def get_exchange_rate(from_currency: str, to_currency: str) -> str:
    rates = {("USD", "EUR"): 0.92, ("EUR", "USD"): 1.09}
    rate = rates.get((from_currency.upper(), to_currency.upper()))
    if rate is None:
        return f"No rate for {from_currency} -> {to_currency}"
    return f"1 {from_currency.upper()} = {rate} {to_currency.upper()}"


def get_current_time() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")


TOOL_FUNCTIONS = {
    "get_stock_price": get_stock_price,
    "get_exchange_rate": get_exchange_rate,
    "get_current_time": get_current_time,
}

TOOLS = [
    {
        "name": "get_stock_price",
        "description": "Get the current price of a stock by its ticker symbol.",
        "input_schema": {
            "type": "object",
            "properties": {
                "ticker": {"type": "string", "description": "Stock ticker, e.g. 'AAPL'"},
                "currency": {"type": "string", "enum": ["USD", "EUR", "GBP"], "description": "Defaults to USD"},
            },
            "required": ["ticker"],
        },
    },
    {
        "name": "get_exchange_rate",
        "description": "Get the exchange rate between two currencies.",
        "input_schema": {
            "type": "object",
            "properties": {
                "from_currency": {"type": "string", "description": "3-letter currency code, e.g. 'USD'"},
                "to_currency": {"type": "string", "description": "3-letter currency code, e.g. 'EUR'"},
            },
            "required": ["from_currency", "to_currency"],
        },
    },
    {
        "name": "get_current_time",
        "description": "Get the current date and time in UTC.",
        "input_schema": {"type": "object", "properties": {}},
    },
]

def run_agent(user_message: str, max_steps: int = 5) -> str:
    messages = [{"role": "user", "content": user_message}]

    for _ in range(max_steps):
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            tools=TOOLS,
            messages=messages,
        )

        if response.stop_reason != "tool_use":
            return response.content[0].text

        messages.append({"role": "assistant", "content": response.content})

        tool_results = []
        for block in response.content:
            if block.type != "tool_use":
                continue
            func = TOOL_FUNCTIONS[block.name]
            try:
                result = func(**block.input)
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": result,
                })
            except Exception as e:
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": str(e),
                    "is_error": True,
                })

        messages.append({"role": "user", "content": tool_results})

    return "Reached max steps without a final answer."


if __name__ == "__main__":
    print(run_agent("What time is it, and how much is AAPL worth in EUR right now?"))

A single call here can trigger Claude to request get_current_time, get_stock_price, and get_exchange_rate — possibly in one parallel batch — then combine all three results into a final answer.

Summary

A tool is name + description + input_schema (JSON Schema)
Tool use is a loop: call → check stop_reason == "tool_use" → execute every tool_use block → send tool_result blocks → repeat
Claude can request multiple tools in one turn — match by tool_use_id, not order
tool_choice: auto, any, a specific tool, or none — forcing a tool is a clean way to get structured output
get_final_message() gives parsed tool_use blocks even when streaming
Set is_error: true on failed tool results
Write clear descriptions, use enums, keep required fields minimal, cap agent loops with a step limit

How to Build a WordPress AI Plugin (Step-by-Step Guide 2026)

Serhii Kalyna — Thu, 11 Jun 2026 08:45:39 +0000

How to Build a WordPress AI Plugin (Step-by-Step Guide 2026)

This guide builds a complete WordPress plugin with AI content generation: settings page, OpenAI API wrapper, secure AJAX handlers, and a Gutenberg sidebar panel.

Plugin Structure

wp-ai-assistant/
├── wp-ai-assistant.php    # Headers + init
├── includes/class-api.php # OpenAI wrapper
├── includes/class-ajax.php# AJAX handlers
└── assets/editor.js       # Gutenberg sidebar

OpenAI Wrapper

class WP_AI_Assistant_API {
    public function generate(string $prompt): string {
        $client = OpenAI::client(get_option('wp_ai_assistant_api_key'));
        $response = $client->chat()->create([
            'model' => 'gpt-4o-mini',
            'messages' => [['role' => 'user', 'content' => $prompt]],
        ]);
        return $response->choices[0]->message->content;
    }
}

Secure AJAX Handler

add_action('wp_ajax_wp_ai_generate', function () {
    check_ajax_referer('wp_ai_assistant_nonce', 'nonce');
    if (!current_user_can('edit_posts')) wp_send_json_error('Forbidden', 403);
    $prompt = sanitize_textarea_field($_POST['prompt'] ?? '');
    try {
        $result = (new WP_AI_Assistant_API())->generate($prompt);
        wp_send_json_success(['content' => $result]);
    } catch (Exception $e) {
        wp_send_json_error($e->getMessage());
    }
});

Gutenberg Sidebar (editor.js)

const { registerPlugin } = wp.plugins;
registerPlugin('wp-ai-assistant', { render: AIAssistantPanel });
// Panel calls admin-ajax.php → inserts generated content as a paragraph block

Security Checklist

check_ajax_referer() on every AJAX action
current_user_can() before processing
sanitize_* on all inputs, esc_* on all outputs
Store API key in wp_options, never in code

Originally published at kalyna.pro

How to Build a Telegram Bot with PHP and AI (2026)

Serhii Kalyna — Thu, 11 Jun 2026 08:44:05 +0000

How to Build a Telegram Bot with PHP and AI (2026)

This guide builds a fully working AI Telegram bot with PHP: webhook setup, message handling, per-user conversation history, AI replies via OpenAI, and inline keyboards.

Requirements

PHP 8.1+, Composer, OpenAI API key, Telegram bot token

Install

composer require openai-php/client guzzlehttp/guzzle

Conversation History (per-user JSON files)

class ConversationHistory {
    public function get(int $userId): array {
        $file = "/tmp/tg_history/{$userId}.json";
        return file_exists($file) ? json_decode(file_get_contents($file), true) :
            [['role' => 'system', 'content' => 'You are a helpful assistant.']];
    }
    public function append(int $userId, string $role, string $content): void {
        $msgs = $this->get($userId);
        $msgs[] = ['role' => $role, 'content' => $content];
        file_put_contents("/tmp/tg_history/{$userId}.json", json_encode($msgs));
    }
}

AI Reply

$client = OpenAI::client(getenv('OPENAI_API_KEY'));
$response = $client->chat()->create([
    'model' => 'gpt-4o-mini',
    'messages' => $this->history->get($userId),
]);
$reply = $response->choices[0]->message->content;
$this->history->append($userId, 'assistant', $reply);

Webhook Entry Point

// webhook.php
$update = json_decode(file_get_contents('php://input'), true);
if ($update) { (new TelegramAIBot())->handle($update); }
http_response_code(200);

Register Webhook

curl -X POST "https://api.telegram.org/bot<TOKEN>/setWebhook" \
  -d "url=https://yourdomain.com/webhook.php"

Key Points

Always return HTTP 200 (or Telegram retries)
Show typing action with sendChatAction before slow AI calls
Commands: /start, /clear, /help
Inline keyboards via reply_markup.inline_keyboard

Originally published at kalyna.pro

How to Integrate ChatGPT with PHP (Complete Guide 2026)

Serhii Kalyna — Thu, 11 Jun 2026 08:42:42 +0000

How to Integrate ChatGPT with PHP (Complete Guide 2026)

This guide covers every approach: official PHP SDK, raw cURL, conversation history, streaming SSE, and a working chat form.

Option A: Official SDK

composer require openai-php/client

$client = OpenAI::client(getenv('OPENAI_API_KEY'));
$response = $client->chat()->create([
    'model' => 'gpt-4o-mini',
    'messages' => [['role' => 'user', 'content' => 'Hello!']],
]);
echo $response->choices[0]->message->content;

Option B: Raw cURL

$ch = curl_init('https://api.openai.com/v1/chat/completions');
curl_setopt_array($ch, [
    CURLOPT_POST => true,
    CURLOPT_POSTFIELDS => json_encode(['model' => 'gpt-4o-mini', 'messages' => [['role' => 'user', 'content' => $msg]]]),
    CURLOPT_RETURNTRANSFER => true,
    CURLOPT_HTTPHEADER => ['Content-Type: application/json', 'Authorization: Bearer ' . $key],
]);
$data = json_decode(curl_exec($ch), true);
echo $data['choices'][0]['message']['content'];

Conversation History

session_start();
$_SESSION['messages'][] = ['role' => 'user', 'content' => $input];
$response = $client->chat()->create(['model' => 'gpt-4o-mini', 'messages' => $_SESSION['messages']]);
$reply = $response->choices[0]->message->content;
$_SESSION['messages'][] = ['role' => 'assistant', 'content' => $reply];

Streaming SSE

header('Content-Type: text/event-stream');
$stream = $client->chat()->createStreamed(['model' => 'gpt-4o-mini', 'messages' => $messages]);
foreach ($stream as $r) {
    $delta = $r->choices[0]->delta->content;
    if ($delta) { echo "data: " . json_encode(['chunk' => $delta]) . "\n\n"; ob_flush(); flush(); }
}
echo "data: [DONE]\n\n";

Error Handling

use OpenAI\Exceptions\ErrorException;
use OpenAI\Exceptions\TransporterException;
try {
    $reply = $client->chat()->create([...])->choices[0]->message->content;
} catch (ErrorException $e) { /* API error */ }
  catch (TransporterException $e) { /* Network error */ }

Originally published at kalyna.pro

Vector Database Tutorial: Build Semantic Search from Scratch (2026)

Serhii Kalyna — Thu, 11 Jun 2026 08:41:14 +0000

Vector Database Tutorial: Build Semantic Search from Scratch (2026)

A vector database stores data as high-dimensional vectors and retrieves by semantic similarity. This tutorial builds a complete pipeline: embeddings, ChromaDB, semantic search, metadata filtering, and a full PDF Q&A example.

Install

pip install chromadb openai

Generate Embeddings

from openai import OpenAI
client = OpenAI()

def embed(text: str) -> list[float]:
    return client.embeddings.create(model="text-embedding-3-small", input=text).data[0].embedding

Set Up ChromaDB

import chromadb
db = chromadb.PersistentClient(path="./chroma_store")
collection = db.get_or_create_collection("documents", metadata={"hnsw:space": "cosine"})

Add Documents

texts = ["Python is readable", "Rust is fast", "Docker containers isolate apps"]
vectors = [embed(t) for t in texts]
collection.add(ids=["1","2","3"], embeddings=vectors, documents=texts)

Semantic Search

def search(query: str, n: int = 3):
    results = collection.query(query_embeddings=[embed(query)], n_results=n)
    return [(doc, round(1-dist, 3)) for doc, dist in
            zip(results["documents"][0], results["distances"][0])]

for doc, score in search("running apps in containers"):
    print(f"[{score}] {doc}")

Filter by Metadata

results = collection.query(
    query_embeddings=[embed("web development")],
    n_results=2,
    where={"language": {"$eq": "python"}},
)

Scaling

ChromaDB — local, great for dev and < 1M vectors
Pinecone — managed, billions of vectors, serverless
pgvector — if you're already on PostgreSQL
Qdrant — high-performance Rust, rich filtering

Originally published at kalyna.pro

How to Build an AI Chatbot with PHP (Step-by-Step, 2026)

Serhii Kalyna — Thu, 11 Jun 2026 08:39:01 +0000

How to Build an AI Chatbot with PHP (Step-by-Step, 2026)

This guide builds a working web AI chatbot with PHP: a backend that calls OpenAI, session-based conversation history, streaming SSE, and a clean HTML/JS frontend.

Requirements

PHP 8.1+, Composer, OpenAI API key

Install

composer require openai-php/client

chat.php — Non-Streaming Backend

<?php
require 'vendor/autoload.php';
session_start();
header('Content-Type: application/json');

$input = json_decode(file_get_contents('php://input'), true)['message'] ?? '';

if (!isset($_SESSION['history'])) {
    $_SESSION['history'] = [['role' => 'system', 'content' => 'You are a helpful assistant.']];
}

$_SESSION['history'][] = ['role' => 'user', 'content' => $input];

$client = OpenAI::client(getenv('OPENAI_API_KEY'));
$response = $client->chat()->create([
    'model' => 'gpt-4o-mini',
    'messages' => array_slice($_SESSION['history'], -20),
]);

$reply = $response->choices[0]->message->content;
$_SESSION['history'][] = ['role' => 'assistant', 'content' => $reply];

echo json_encode(['reply' => $reply]);

stream.php — Streaming SSE Backend

<?php
require 'vendor/autoload.php';
session_start();
header('Content-Type: text/event-stream');
header('Cache-Control: no-cache');

$input = json_decode(file_get_contents('php://input'), true)['message'] ?? '';
$_SESSION['history'][] = ['role' => 'user', 'content' => $input];

$stream = OpenAI::client(getenv('OPENAI_API_KEY'))->chat()->createStreamed([
    'model' => 'gpt-4o-mini',
    'messages' => $_SESSION['history'],
]);

$full = '';
foreach ($stream as $response) {
    $delta = $response->choices[0]->delta->content;
    if ($delta !== null) {
        $full .= $delta;
        echo "data: " . json_encode(['chunk' => $delta]) . "\n\n";
        ob_flush(); flush();
    }
}
$_SESSION['history'][] = ['role' => 'assistant', 'content' => $full];
echo "data: [DONE]\n\n";

Frontend (EventSource / fetch streaming)

const res = await fetch('stream.php', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ message: text }),
});

const reader = res.body.getReader();
const decoder = new TextDecoder();
let buffer = '';

while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    buffer += decoder.decode(value, { stream: true });
    for (const line of buffer.split('\n')) {
        if (!line.startsWith('data: ')) continue;
        const data = line.slice(6);
        if (data === '[DONE]') return;
        const { chunk } = JSON.parse(data);
        outputEl.textContent += chunk;
    }
}

Production Checklist

Set OPENAI_API_KEY as server env var (never hardcode)
Use HTTPS for SSE + sessions
Trim $_SESSION['history'] to avoid token overflow
Add per-user rate limiting to control costs
Catch ErrorException + TransporterException

Originally published at kalyna.pro

OpenAI PHP SDK: Complete Guide with Code Examples (2026)

Serhii Kalyna — Thu, 11 Jun 2026 08:37:11 +0000

OpenAI PHP SDK: Complete Guide with Code Examples (2026)

The openai-php/client is the official PHP library for the OpenAI API. This guide covers every major feature: chat completions, streaming, function calling, embeddings, image generation, audio, and error handling.

Installation

composer require openai-php/client

Initialization

$client = OpenAI::client(getenv('OPENAI_API_KEY'));

Chat Completions

$response = $client->chat()->create([
    'model' => 'gpt-4o-mini',
    'messages' => [
        ['role' => 'system', 'content' => 'You are a helpful assistant.'],
        ['role' => 'user',   'content' => 'What is the capital of France?'],
    ],
]);
echo $response->choices[0]->message->content; // Paris

Streaming

$stream = $client->chat()->createStreamed([
    'model' => 'gpt-4o-mini',
    'messages' => [['role' => 'user', 'content' => 'Write a poem about PHP.']],
]);
foreach ($stream as $response) {
    echo $response->choices[0]->delta->content;
}

Function Calling

$response = $client->chat()->create([
    'model' => 'gpt-4o-mini',
    'messages' => [['role' => 'user', 'content' => "What's the weather in Kyiv?"]],
    'tools' => [[
        'type' => 'function',
        'function' => ['name' => 'get_weather', 'description' => 'Get city weather',
            'parameters' => ['type' => 'object',
                'properties' => ['city' => ['type' => 'string']], 'required' => ['city']]],
    ]],
]);
$toolCalls = $response->choices[0]->message->toolCalls;

Embeddings

$response = $client->embeddings()->create([
    'model' => 'text-embedding-3-small',
    'input' => 'Laravel is a PHP framework.',
]);
$vector = $response->embeddings[0]->embedding; // 1536-dim float array

Image Generation

$response = $client->images()->create([
    'model' => 'dall-e-3', 'prompt' => 'A futuristic PHP elephant',
    'n' => 1, 'size' => '1024x1024',
]);
echo $response->data[0]->url;

Error Handling

use OpenAI\Exceptions\ErrorException;
use OpenAI\Exceptions\TransporterException;

try {
    $reply = $client->chat()->create([...]);
} catch (ErrorException $e) {
    // API error (rate limit, auth, etc.)
} catch (TransporterException $e) {
    // Network error
}

Summary

chat()->create() for completions, createStreamed() for streaming
embeddings()->create() for vectors, images()->create() for DALL·E
audio()->transcribe() for Whisper, moderations()->create() for safety
Catch ErrorException + TransporterException in production

Originally published at kalyna.pro

How to Integrate OpenAI API with Laravel (Complete Guide 2026)

Serhii Kalyna — Thu, 11 Jun 2026 08:35:08 +0000

How to Integrate OpenAI API with Laravel (Complete Guide 2026)

Laravel is the most popular PHP framework, and adding OpenAI to it is straightforward with the official openai-php/client library. This guide covers installation, a service class, a chat controller, streaming, error handling, and testing.

Requirements

Laravel 10 or 11, PHP 8.1+, Composer, OpenAI API key

Step 1: Install

composer require openai-php/laravel
php artisan vendor:publish --provider="OpenAI\Laravel\ServiceProvider"

Step 2: .env

OPENAI_API_KEY=sk-proj-...

Step 3: OpenAIService

namespace App\Services;
use OpenAI\Laravel\Facades\OpenAI;

class OpenAIService
{
    public function chat(array $messages, string $model = 'gpt-4o-mini'): string
    {
        $response = OpenAI::chat()->create(['model' => $model, 'messages' => $messages]);
        return $response->choices[0]->message->content;
    }
}

Step 4: ChatController

public function send(Request $request): JsonResponse
{
    $messages = $request->input('history', []);
    $messages[] = ['role' => 'user', 'content' => $request->input('message')];
    $reply = $this->openAI->chat($messages);
    return response()->json(['reply' => $reply, 'messages' => [...$messages,
        ['role' => 'assistant', 'content' => $reply]]]);
}

Step 5: Streaming SSE

return response()->stream(function () use ($messages) {
    foreach ($this->openAI->stream($messages) as $chunk) {
        echo "data: " . json_encode(['chunk' => $chunk]) . "\n\n";
        ob_flush(); flush();
    }
    echo "data: [DONE]\n\n";
}, 200, ['Content-Type' => 'text/event-stream', 'Cache-Control' => 'no-cache']);

Error Handling

use OpenAI\Exceptions\ErrorException;
use OpenAI\Exceptions\TransporterException;

try {
    $reply = $this->openAI->chat($messages);
} catch (ErrorException $e) {
    return response()->json(['error' => 'AI service error.'], 503);
} catch (TransporterException $e) {
    return response()->json(['error' => 'Connection failed.'], 503);
}

Testing

OpenAI::fake([CreateResponse::fake(['choices' => [['message' => ['content' => 'Hi!']]]])]);
$reply = app(OpenAIService::class)->chat([['role' => 'user', 'content' => 'Hello']]);
$this->assertEquals('Hi!', $reply);

Summary

composer require openai-php/laravel + OPENAI_API_KEY in .env
Service class keeps controllers clean; stream with response()->stream() + SSE
OpenAI::fake() for unit tests without hitting the real API

Originally published at kalyna.pro

How to Build a Telegram Bot with Python and AI (2026)

Serhii Kalyna — Thu, 11 Jun 2026 08:31:17 +0000

A Telegram bot is one of the most practical ways to deploy an AI assistant — your users already have Telegram, there's no app to install, and the Bot API is simple. This guide builds a fully functional AI bot with python-telegram-bot and the Claude API, including commands, inline keyboards, conversation history, error handling, and systemd deployment.

Prerequisites

Python 3.10+
Telegram Bot Token from @botfather
Anthropic API key (ANTHROPIC_API_KEY)

Step 1: Create the Bot

Open Telegram, find @botfather, send /newbot, and save the token.

Step 2: Install Dependencies

pip install python-telegram-bot anthropic

Step 3: Basic Bot Structure

import os
import anthropic
from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes

TELEGRAM_TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]
claude = anthropic.Anthropic()

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    await update.message.reply_text("Hi! I'm an AI assistant. Send me a message!")

def main() -> None:
    app = Application.builder().token(TELEGRAM_TOKEN).build()
    app.add_handler(CommandHandler("start", start))
    app.run_polling()

if __name__ == "__main__":
    main()

Step 4: Conversation History

MAX_HISTORY = 20

def get_history(context):
    return context.chat_data.setdefault("history", [])

async def chat(update: Update, context: ContextTypes.DEFAULT_TYPE):
    history = get_history(context)
    history.append({"role": "user", "content": update.message.text})
    if len(history) > MAX_HISTORY:
        del history[:len(history) - MAX_HISTORY]

    response = claude.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system="You are a helpful AI assistant in Telegram. Be concise.",
        messages=history,
    )
    answer = response.content[0].text
    history.append({"role": "assistant", "content": answer})
    await update.message.reply_text(answer)

Step 5: Custom Commands

async def summarize(update: Update, context: ContextTypes.DEFAULT_TYPE):
    if not context.args:
        await update.message.reply_text("Usage: /summarize <text>")
        return
    text = " ".join(context.args)
    resp = claude.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=512,
        messages=[{"role": "user", "content": f"Summarize in 3 bullet points:\n{text}"}],
    )
    await update.message.reply_text(resp.content[0].text)

Step 6: Inline Keyboards

from telegram import InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import CallbackQueryHandler

async def mode(update: Update, context: ContextTypes.DEFAULT_TYPE):
    keyboard = [[
        InlineKeyboardButton("Concise", callback_data="mode_concise"),
        InlineKeyboardButton("Detailed", callback_data="mode_detailed"),
    ]]
    await update.message.reply_text("Choose mode:", reply_markup=InlineKeyboardMarkup(keyboard))

async def mode_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
    query = update.callback_query
    await query.answer()
    context.chat_data["mode"] = query.data.replace("mode_", "")
    await query.edit_message_text(f"Mode set: {context.chat_data['mode']}")

Step 7: Deploy with systemd

# /etc/systemd/system/telegram-ai-bot.service
[Unit]
Description=Telegram AI Bot
After=network.target

[Service]
User=ubuntu
WorkingDirectory=/home/ubuntu/telegram-bot
ExecStart=/home/ubuntu/telegram-bot/venv/bin/python bot.py
Environment=TELEGRAM_BOT_TOKEN=7123...
Environment=ANTHROPIC_API_KEY=sk-ant-...
Restart=always

[Install]
WantedBy=multi-user.target

sudo systemctl enable --now telegram-ai-bot

Summary

You built a production-ready AI Telegram bot with conversation history, custom commands, inline keyboards, and systemd deployment.

Originally published at kalyna.pro

How to Use OpenAI API with PHP (Complete Guide, 2026)

Serhii Kalyna — Fri, 05 Jun 2026 11:42:08 +0000

How to Use OpenAI API with PHP (Complete Guide, 2026)

PHP powers a large part of the web, but most AI tutorials focus on Python. This guide fills that gap: a complete OpenAI PHP tutorial with working code you can use in any PHP project or Laravel app.

Installation

composer require openai-php/client

Basic Setup

$client = OpenAI::client($_ENV['OPENAI_API_KEY']);

Chat Completions

$response = $client->chat()->create([
    'model' => 'gpt-4o',
    'messages' => [
        ['role' => 'user', 'content' => 'Explain what a PHP trait is.'],
    ],
    'max_tokens' => 256,
]);
echo $response->choices[0]->message->content;

Key Features

Streaming — SSE responses with createStreamed()
Function calling — expose PHP functions as tools
Vision — pass images as URL or base64
Laravel — register as singleton in service provider

Cost Tips

Use gpt-4o-mini for simple tasks (17x cheaper than gpt-4o)
Set max_tokens explicitly
Cache responses in Redis for repeated prompts

Originally published at kalyna.pro