Perplexity API Python Tutorial

# 1. Create and activate a virtual environment
python -m venv .venv
source .venv/bin/activate        # Windows: .venv\Scripts\activate

# 2. Install the official SDK (and dotenv for local keys)
pip install perplexityai python-dotenv

# .env  (add this file to .gitignore)
PERPLEXITY_API_KEY=your_api_key_here

from dotenv import load_dotenv
from perplexity import Perplexity

load_dotenv()           # loads PERPLEXITY_API_KEY from .env
client = Perplexity()   # reads the key from the environment

response = client.responses.create(
    preset="fast",
    input="What are the latest developments in AI agents?",
)

print(response.output_text)

# Use a preset (recommended): Perplexity picks the model and tools
response = client.responses.create(preset="low", input="Compare solar and wind costs in 2026")

# Or choose a specific model yourself
response = client.responses.create(
    model="openai/gpt-5.6-sol",
    input="Explain supervised learning in simple terms",
)
print(response.output_text)

response = client.responses.create(
    preset="fast",
    instructions="You are a friendly tutor. Answer in under 120 words, in simple English.",
    input="How do vaccines train the immune system?",
)
print(response.output_text)

from perplexity import Perplexity

client = Perplexity()
history = []

while True:
    question = input("You: ").strip()
    if question.lower() in {"quit", "exit"}:
        break
    history.append({"role": "user", "content": question})

    response = client.responses.create(preset="fast", input=history)
    answer = response.output_text
    print(f"Perplexity: {answer}\n")

    history.append({"role": "assistant", "content": answer})

stream = client.responses.create(
    preset="fast",
    input="Summarize today's top technology news in 5 bullets",
    stream=True,
)

for event in stream:
    if event.type == "response.output_text.delta":
        print(event.delta, end="", flush=True)
    elif event.type == "response.completed":
        print(f"\n\nUsage: {event.response.usage}")

response = client.responses.create(
    preset="fast",
    input="How much of the Amazon rainforest has been lost?",
)

print(response.output_text, "\n")

# Sources arrive as output items of type "search_results"
for item in response.output:
    if item.type == "search_results":
        for result in item.results:
            print(f"[{result.id}] {result.title}")
            print(f"    {result.url}  ({result.date})")

response = client.responses.create(
    preset="fast",
    input="Latest renewable energy policy updates",
    tools=[
        {
            "type": "web_search",
            "filters": {
                "search_domain_filter": ["iea.org", "energy.gov"],
                "search_recency_filter": "month",
            },
        }
    ],
)
print(response.output_text)

import json

response = client.responses.create(
    preset="low",
    input="List the 3 largest countries by area with their area in km2",
    response_format={
        "type": "json_schema",
        "json_schema": {
            "name": "countries",
            "schema": {
                "type": "object",
                "properties": {
                    "countries": {
                        "type": "array",
                        "items": {
                            "type": "object",
                            "properties": {
                                "name": {"type": "string"},
                                "area_km2": {"type": "number"},
                            },
                            "required": ["name", "area_km2"],
                        },
                    }
                },
                "required": ["countries"],
            },
        },
    },
)

data = json.loads(response.output_text)
for c in data["countries"]:
    print(c["name"], c["area_km2"])

import time

response = client.responses.create(
    preset="high",
    input="Write a detailed report on battery recycling technologies, with sources",
    background=True,
)

while response.status in ("queued", "in_progress"):
    time.sleep(2)
    response = client.responses.retrieve(response.id)

print(response.status)
print(response.output_text)

search = client.search.create(
    query="EV battery recycling 2026",
    max_results=5,
    search_recency_filter="month",
)

for result in search.results:
    print(f"{result.title}: {result.url}")

import asyncio
from perplexity import AsyncPerplexity

client = AsyncPerplexity()

async def ask(question: str) -> str:
    response = await client.responses.create(preset="fast", input=question)
    return response.output_text

async def main():
    questions = [
        "What is the capital of Australia?",
        "Who discovered penicillin?",
        "How tall is Mount Everest?",
    ]
    answers = await asyncio.gather(*(ask(q) for q in questions))
    for q, a in zip(questions, answers):
        print(f"Q: {q}\nA: {a}\n")

asyncio.run(main())

import random
import time

def ask_with_retry(question: str, attempts: int = 5) -> str:
    for attempt in range(attempts):
        try:
            response = client.responses.create(preset="fast", input=question)
            return response.output_text
        except Exception as error:
            status = getattr(error, "status_code", None)
            retryable = status == 429 or (status is not None and status >= 500)
            if not retryable or attempt == attempts - 1:
                raise
            wait = (2 ** attempt) + random.random()
            print(f"Got {status}, retrying in {wait:.1f}s...")
            time.sleep(wait)