Search the docs, videos, blog, apps, teams, skills and release notes.
At a glance
What it is
LLM API integration covering OpenAI, Anthropic, and Google API patterns, streaming responses, function calling and tool use, token counting, cost optimization, fallback strategies, rate limit handling, response caching, and multi-model routing.
A skill is a written procedure a bot loads when a job calls for it. This one is a single file, SKILL.md, and the whole file is on this page.
The whole file, exactly as it downloads. Read it before you rely on it.
SKILL.md
438 lines · 1,320 words
Integrating LLMs into production applications requires robust API handling, cost management, and reliability patterns. This skill covers practical integration with major LLM providers, including streaming, tool use, error handling, and multi-model architectures for production systems.
LLM API integration covering OpenAI, Anthropic, and Google API patterns, streaming responses, function calling and tool use, token counting, cost optimization, fallback strategies, rate limit handling, response caching, and multi-model routing.
Nothing. It ships with Murage, which is open source. Your bots run on the AI plan you already have.
Yes. Open it in Settings → Skills to read, edit or duplicate it. Skill Guard scans it again after every edit.
Download the app, connect the AI you already pay for, and tell your Chief of Staff what needs doing. Plan on about ten minutes from install to a working team.
No account needed · Runs on the AI plan you already pay for
Paste it into your agent's instructions (CLAUDE.md, AGENTS.md or a custom GPT), or add it to a bot in Murage.
Ask Ember for the bot, then switch the skill on in its settings.
Step 1
Send your Chief of Staff this message. It proposes the bot and its role first, and creates it when you say go.
Send this to Ember
“Create a bot that uses the LLM Integrator skill, and show me the plan first.”
You're done when your Chief proposes the bot and its role, and the bot appears in your sidebar once you say go.
Step 2
You're done when you see the bot’s skills and the library.
Step 3
You're done when the skill shows as on for that bot.
Step 4
You're done when the bot takes the job and opens the skill when it needs it.
Prefer to do it yourself? Choose + in the sidebar, then New Bot.Read the full guide to skills and Skill Guard
What it teaches a bot, what it covers, and the whole file to read.
What it covers
Overview
Provider API Patterns
Streaming Responses
Function Calling / Tool Use
Token Counting and Cost Management
Rate Limit Handling
Fallback Strategies
Response Caching
Multi-Model Routing
Production Architecture
Checklist
When to Use
Output Format
Example
Edge Cases
from openai import OpenAI, AsyncOpenAI
client = OpenAI(api_key=environment-variables["OPENAI_API_KEY"])
# Basic completion
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Explain quantum computing."},
],
# ... (condensed) ...
for p in prompts
]
responses = await asyncio.gather(*tasks)
return [r.choices[0].message.content for r in responses]from anthropic import Anthropic, AsyncAnthropic
client = Anthropic(api_key=environment-variables["ANTHROPIC_API_KEY"])
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1024,
system="You are an expert software architect.",
messages=[
{"role": "user", "content": "Design a microservices architecture for e-commerce."},
],
# ... (condensed) ...
if block.type == "thinking":
print(f"Thinking: {block.thinking}")
elif block.type == "text":
print(f"Answer: {block.text}")import google.generativeai as genai
genai.configure(api_key=environment-variables["GOOGLE_API_KEY"])
model = genai.GenerativeModel("gemini-2.0-flash")
response = model.generate_content("Explain neural networks.")
print(response.text)
# Multi-modal input
import PIL.Image
image = PIL.Image.open("diagram.png")
response = model.generate_content(["Describe this diagram:", image])
print(response.text)def stream_openai(prompt: str):
"""Stream response tokens as they are generated."""
stream = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
stream=True,
)
full_response = ""
for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
full_response += delta
print(delta, end="", flush=True)
return full_responsedef stream_anthropic(prompt: str):
"""Stream with Anthropic's event-based API."""
with client.messages.stream(
model="claude-sonnet-4-20250514",
max_tokens=1024,
messages=[{"role": "user", "content": prompt}],
) as stream:
full_response = ""
for text in stream.text_stream:
full_response += text
print(text, end="", flush=True)
return full_responsefrom fastapi import FastAPI
from fastapi.responses import StreamingResponse
app = FastAPI()
@app.get("/chat/stream")
async def chat_stream(prompt: str):
async def event_generator():
async with async_client.messages.stream(
model="claude-sonnet-4-20250514",
max_tokens=1024,
# ... (condensed) ...
return StreamingResponse(
event_generator(),
media_type="text/event-stream",
)tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get current weather for a location",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
# ... (condensed) ...
"content": json.dumps(result),
})
else:
return message.contenttools = [
{
"name": "get_weather",
"description": "Get current weather for a location",
"input_schema": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "City and state"
}
# ... (condensed) ...
messages.append({"role": "assistant", "content": response.content})
messages.append({"role": "user", "content": tool_results})
else:
return next(b.text for b in response.content if b.type == "text")import tiktoken
class TokenCounter:
"""Count tokens for OpenAI models."""
COST_PER_1M = {
"gpt-4o": {"input": 2.50, "output": 10.00},
"gpt-4o-mini": {"input": 0.15, "output": 0.60},
"gpt-4-turbo": {"input": 10.00, "output": 30.00},
}
# ... (condensed) ...
return (
input_tokens * costs["input"] / 1_000_000 +
output_tokens * costs["output"] / 1_000_000
)class CostOptimizedClient:
"""Route requests to minimize cost while maintaining quality."""
def __init__(self):
self.openai = OpenAI()
self.anthropic = Anthropic()
self.token_counter = TokenCounter()
def complete(
self,
prompt: str,
# ... (condensed) ...
for kw in ["architecture", "complex", "multi-step"]
):
return "complex"
return "medium"import time
from tenacity import (
retry, stop_after_attempt, wait_exponential,
retry_if_exception_type,
)
from openai import RateLimitError, APITimeoutError, InternalServerError
@retry(
retry=retry_if_exception_type((RateLimitError, APITimeoutError, InternalServerError)),
wait=wait_exponential(multiplier=1, min=1, max=60),
stop=stop_after_attempt(6),
# ... (condensed) ...
)
def resilient_completion(client, **kwargs):
"""LLM call with automatic retry on transient errors."""
return client.chat.completions.create(**kwargs)import asyncio
import time
class TokenBucketRateLimiter:
"""Rate limiter for API calls respecting tokens-per-minute limits."""
def __init__(self, tokens_per_minute: int, requests_per_minute: int):
self.tpm = tokens_per_minute
self.rpm = requests_per_minute
self.token_bucket = tokens_per_minute
self.request_bucket = requests_per_minute
# ... (condensed) ...
self.rpm,
self.request_bucket + elapsed * self.rpm / 60
)
self.last_refill = nowclass LLMRouter:
"""Route LLM calls with automatic fallback across providers."""
def __init__(self):
self.providers = [
{
"name": "openai",
"client": OpenAI(),
"model": "gpt-4o",
"priority": 1,
"healthy": True,
# ... (condensed) ...
max_tokens=kwargs.get("max_tokens", 1024),
messages=messages,
)
return resp.content[0].textimport hashlib
import json
import redis
class LLMCache:
"""Cache LLM responses with exact matching."""
def __init__(self, redis_url: str, ttl: int = 86400):
self.redis = redis.from_url(redis_url)
self.ttl = ttl
# ... (condensed) ...
result = response.choices[0].message.content
self.set(messages, model, result)
return resultclass SmartRouter:
"""Route requests to optimal model based on task analysis."""
MODEL_CONFIGS = {
"fast_cheap": {
"provider": "openai",
"model": "gpt-4o-mini",
"max_tokens": 1024,
"cost_input": 0.15,
"cost_output": 0.60,
},
# ... (condensed) ...
)
classification = resp.choices[0].message.content.strip().lower()
mapping = {"simple": "fast_cheap", "medium": "balanced", "complex": "max_quality"}
return mapping.get(classification, "balanced")class LLMService:
"""Production-ready LLM service with all reliability patterns."""
def __init__(self, config: dict):
self.router = SmartRouter()
self.cache = LLMCache(config["redis_url"])
self.rate_limiter = TokenBucketRateLimiter(
tokens_per_minute=config["tpm"],
requests_per_minute=config["rpm"],
)
self.fallback = LLMRouter()
# ... (condensed) ...
self.metrics.record("latency", latency)
self.metrics.record("model_used", config["model"])
return resultSet up API clients for all required providers with proper auth
Implement streaming for user-facing interactions
Add function calling / tool use for agentic workflows
Build token counting for cost estimation before calls
Configure retry logic with exponential backoff for transient errors
Implement multi-provider fallback chain
Add response caching for repeated queries
Set up rate limiting to respect API quotas
Build cost monitoring and alerting
Route requests to optimal model based on task complexity
Log all API calls for debugging and cost analysis
Test failover paths regularly
Use this skill when:
Designing or implementing llm integrator solutions
Reviewing or improving existing llm integrator approaches
Making architectural or implementation decisions about llm integrator
Learning llm integrator patterns and best practices
Troubleshooting llm integrator-related issues
Do NOT use this skill when:
The question is about a fundamentally different technology domain
A more specific sibling skill covers the exact topic needed
The user needs a complete hands-on tutorial rather than expert guidance
# Llm Integrator Analysis
## Context Assessment
[Situation summary and constraints]
## Recommended Approach
[Primary recommendation with rationale]
## Implementation Steps
1. [Step with specific details]
2. [Step with specific details]
3. [Step with specific details]
## Trade-offs and Considerations
- [Key trade-off 1]
- [Key trade-off 2]
## Next Steps
- [Immediate action item]
- [Follow-up action item]Input: "Help me implement llm integrator for a medium-scale production application"
Output: A structured analysis covering current state assessment, recommended llm integrator approach with specific patterns, implementation roadmap with milestones, and risk mitigation strategies tailored to the application scale and constraints.
Legacy system integration: When llm integrator must coexist with legacy approaches, provide a gradual migration path rather than a complete rewrite
Scale mismatch: When the solution complexity exceeds the project scale, recommend a simpler approach and note when to revisit
Team skill gaps: When the team lacks experience with the recommended approach, include learning resources and simpler alternatives
Conflicting requirements: When constraints conflict (e.g., performance vs. maintainability), explicitly state the trade-off and recommend based on stated priorities
SKILL.md
438 lines · 1,320 words
The text of SKILL.md is shown as it downloads, under the Apache-2.0 license, by Ferrox Labs. The license is the one in the file's own frontmatter.
Your new bot
LLM Integrator bot
Waiting in your sidebar once you say go.
No account needed · Runs on the AI plan you already pay for