LLM API for Developers — Complete Integration Guide 2026 | APIMaster.ai
Developer guide to LLM APIs: authentication, streaming, function calling, embeddings, RAG, async patterns, and cost management. Works with Claude, GPT, and DeepSeek via APIMaster.
LLM API for Developers: Complete Integration Guide
This guide covers everything a developer needs to integrate LLM APIs into production applications: authentication, streaming, tool use, embeddings, RAG patterns, and cost management. All examples use the OpenAI-compatible format and work with APIMaster.ai.
Setup
from openai import OpenAI
client = OpenAI(
api_key="YOUR_APIMASTER_KEY",
base_url="https://apimaster.ai/v1",
)
Core Patterns
1. Basic Chat Completion
def ask(prompt: str, model: str = "claude-sonnet-4-6") -> str:
response = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}],
)
return response.choices[0].message.content
2. System Prompt + Conversation
class Conversation:
def __init__(self, system: str, model: str = "claude-sonnet-4-6"):
self.model = model
self.messages = [{"role": "system", "content": system}]
def send(self, user_msg: str) -> str:
self.messages.append({"role": "user", "content": user_msg})
resp = client.chat.completions.create(
model=self.model,
messages=self.messages,
)
reply = resp.choices[0].message.content
self.messages.append({"role": "assistant", "content": reply})
return reply
bot = Conversation("You are an expert Python developer.")
print(bot.send("What is the GIL?"))
print(bot.send("How do I work around it?"))
3. Streaming
def stream(prompt: str, model: str = "gpt-4o"):
with client.chat.completions.stream(
model=model,
messages=[{"role": "user", "content": prompt}],
) as s:
for text in s.text_stream:
yield text
for chunk in stream("Explain async/await in Python"):
print(chunk, end="", flush=True)
4. Structured Output
from pydantic import BaseModel
from typing import List
class ExtractedData(BaseModel):
entities: List[str]
sentiment: str
summary: str
import json
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": f"Extract data and return JSON matching this schema: {ExtractedData.schema()}"},
{"role": "user", "content": "Apple reported record revenue. CEO Tim Cook called it exceptional."},
],
response_format={"type": "json_object"},
)
data = ExtractedData(**json.loads(response.choices[0].message.content))
print(data.entities) # ["Apple", "Tim Cook"]
print(data.sentiment) # "positive"
5. Tool Use / Function Calling
import json
tools = [
{
"type": "function",
"function": {
"name": "execute_sql",
"description": "Run a read-only SQL query",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string"},
"database": {"type": "string", "enum": ["users", "orders", "products"]},
},
"required": ["query", "database"],
},
},
}
]
def handle_tool_call(tool_name: str, args: dict) -> str:
# Your implementation
return json.dumps({"result": "mock data"})
def agent_loop(user_msg: str):
messages = [{"role": "user", "content": user_msg}]
while True:
resp = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
)
if resp.choices[0].finish_reason != "tool_calls":
return resp.choices[0].message.content
# Process tool calls
messages.append(resp.choices[0].message)
for tc in resp.choices[0].message.tool_calls:
result = handle_tool_call(tc.function.name, json.loads(tc.function.arguments))
messages.append({"role": "tool", "tool_call_id": tc.id, "content": result})
6. Embeddings
def embed(texts: list[str]) -> list[list[float]]:
response = client.embeddings.create(
model="text-embedding-3-small",
input=texts,
)
return [item.embedding for item in response.data]
# Semantic similarity
import numpy as np
def cosine_similarity(a, b):
return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
vecs = embed(["Python is great", "I love Python", "Java is verbose"])
print(cosine_similarity(vecs[0], vecs[1])) # High: ~0.95
print(cosine_similarity(vecs[0], vecs[2])) # Lower: ~0.70
7. RAG (Retrieval-Augmented Generation)
from typing import List
def rag_query(user_question: str, knowledge_base: List[str]) -> str:
# Step 1: Embed the question
q_embedding = embed([user_question])[0]
doc_embeddings = embed(knowledge_base)
# Step 2: Find most relevant docs
similarities = [cosine_similarity(q_embedding, d) for d in doc_embeddings]
top_indices = sorted(range(len(similarities)), key=lambda i: similarities[i], reverse=True)[:3]
context = "\n\n".join(knowledge_base[i] for i in top_indices)
# Step 3: Generate answer with context
response = client.chat.completions.create(
model="claude-sonnet-4-6",
messages=[
{"role": "system", "content": f"Answer using only this context:\n\n{context}"},
{"role": "user", "content": user_question},
],
)
return response.choices[0].message.content
8. Async for High Throughput
import asyncio
from openai import AsyncOpenAI
async_client = AsyncOpenAI(
api_key="YOUR_APIMASTER_KEY",
base_url="https://apimaster.ai/v1",
)
async def process_batch(prompts: list[str]) -> list[str]:
tasks = [
async_client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": p}],
max_tokens=100,
)
for p in prompts
]
responses = await asyncio.gather(*tasks)
return [r.choices[0].message.content for r in responses]
# Process 50 prompts concurrently
results = asyncio.run(process_batch(my_prompts))
Production Checklist
- API keys in environment variables, not source code
- Retry logic with exponential backoff for 429/500 errors
-
max_tokensset to prevent runaway costs - Streaming for user-facing responses >2 seconds
- Request logging with token counts for cost tracking
- Rate limiter to stay within provider limits
Choose the Right Model
| Use Case | Model | Cost Tier |
|---|---|---|
| Prototyping | deepseek-v4 or gpt-4o-mini | Very low |
| Production chatbot | claude-haiku-4-5 | Low |
| Code assistant | deepseek-v4 or claude-sonnet-4-6 | Low–medium |
| Complex analysis | claude-sonnet-4-6 | Medium |
| Research/reasoning | claude-opus-4-8 or o3 | High |