API Reference

Pipeline Functions

`load(path) → Prompty`

Parse a .prompty file into a typed Prompty object.

Python
TypeScript

agent = prompty.load("chat.prompty")
print(agent.name)        # "chat"
print(agent.model.id)    # "gpt-4o"

import { load } from "@prompty/core";
const agent = await load("chat.prompty");

`render(agent, inputs) → str`

Render the template with inputs. Returns the raw rendered string before parsing into messages.

Python
TypeScript

rendered = prompty.render(agent, inputs={"q": "Hi"})
# "system:\nYou are helpful.\n\nuser:\nHi"

import { render } from "@prompty/core";
const rendered = await render(agent, { q: "Hi" });

`parse(agent, rendered) → list[Message]`

Parse a rendered string into structured messages.

Python
TypeScript

messages = prompty.parse(agent, rendered)
# [Message(role="system", ...), Message(role="user", ...)]

import { parse } from "@prompty/core";
const messages = await parse(agent, rendered);

`prepare(agent, inputs) → list[Message]`

Composite: render + parse + thread expansion. Returns wire-ready messages.

Python
TypeScript

messages = prompty.prepare(agent, inputs={"q": "Hi"})

import { prepare } from "@prompty/core";
const messages = await prepare(agent, { q: "Hi" });

`run(agent, messages) → result`

Composite: call the LLM executor then process the response.

Python
TypeScript

result = prompty.run(agent, messages)
# "Hello! How can I help you?"

# Pass raw=True to get the raw SDK response
response = prompty.run(agent, messages, raw=True)

import { run } from "@prompty/core";
const result = await run(agent, messages);

`process(agent, response) → result`

Extract clean content from a raw LLM response.

result = prompty.process(agent, response)

`execute(path_or_agent, inputs) → result`

Top-level orchestrator: load → prepare → run.

Python
TypeScript

result = prompty.execute("chat.prompty", inputs={"q": "Hi"})

import { execute } from "@prompty/core";
const result = await execute("chat.prompty", { q: "Hi" });

`validate_inputs(agent, inputs)`

Check that all required inputs (those without defaults) are provided. Raises ValueError if any are missing.

prompty.validate_inputs(agent, {"name": "Jane"})
# Raises ValueError if 'name' isn't in the schema or required fields are missing

`headless(api, content, model, ...)`

Create a Prompty programmatically without a .prompty file:

agent = prompty.headless(
    api="embedding",
    content="Hello world",
    model="text-embedding-3-small",
    provider="openai",
    connection={"kind": "key", "apiKey": "sk-..."},
)
result = prompty.run(agent, agent.metadata["content"])

Async Variants

Every pipeline function has an async counterpart (Python):

Sync	Async
`load()`	`load_async()`
`render()`	`render_async()`
`parse()`	`parse_async()`
`prepare()`	`prepare_async()`
`run()`	`run_async()`
`execute()`	`execute_async()`
`process()`	`process_async()`

TypeScript functions are async by default (return Promise).

Agent Mode

`execute_agent(path_or_agent, inputs, tools, ...)`

Run a prompt with an automatic tool-calling loop.

Python
TypeScript

def get_weather(location: str) -> str:
    return f"72°F and sunny in {location}"

result = prompty.execute_agent(
    "agent.prompty",
    inputs={"question": "Weather in Seattle?"},
    tools={"get_weather": get_weather},
    max_iterations=10,
)

import { executeAgent } from "@prompty/core";

function getWeather(location: string): string {
  return `72°F and sunny in ${location}`;
}

const result = await executeAgent(
  "agent.prompty",
  { question: "Weather in Seattle?" },
  { get_weather: getWeather },
);

Parameters:

Parameter	Type	Description
`prompt`	`str \| Prompty`	Path or loaded agent
`inputs`	`dict`	Template inputs
`tools`	`dict[str, Callable]`	Tool function implementations
`max_iterations`	`int`	Loop limit (default 10)
`raw`	`bool`	Skip processing

Error handling:

Bad JSON in tool args → error sent to model for retry
Tool exception → error string sent to model
Missing tool → error message, no crash
Max iterations exceeded → ValueError

`execute_agent_async(...)`

Async variant. Async tool functions are properly awaited.

Connection Registry

`register_connection(name, client=...)`

from openai import AzureOpenAI
from azure.identity import DefaultAzureCredential, get_bearer_token_provider

client = AzureOpenAI(
    azure_endpoint=os.environ["AZURE_ENDPOINT"],
    azure_ad_token_provider=get_bearer_token_provider(
        DefaultAzureCredential(),
        "https://cognitiveservices.azure.com/.default",
    ),
)
prompty.register_connection("my-foundry", client=client)

Then in your .prompty file:

model:
  id: gpt-4o
  provider: foundry
  connection:
    kind: reference
    name: my-foundry

`get_connection(name) → client`

Look up a registered connection. Raises ValueError if not found.

`clear_connections()`

Remove all registered connections. Useful in tests.

Streaming

Set stream: true in model options:

Python
TypeScript

agent = prompty.load("chat.prompty")
messages = prompty.prepare(agent, inputs={...})
agent.model.options.additionalProperties = {"stream": True}

response = prompty.run(agent, messages, raw=True)
for chunk in prompty.process(agent, response):
    print(chunk, end="", flush=True)

agent.model.options.additionalProperties = { stream: true };
const response = await run(agent, messages, { raw: true });
for await (const chunk of process(agent, response)) {
  process.stdout.write(chunk);
}

Structured Output

Define outputs to get JSON-parsed results:

outputs:
  - name: city
    kind: string
  - name: temp
    kind: integer

The processor automatically JSON-parses the response when outputs is present.

Tracing

from prompty import Tracer, PromptyTracer, trace

Tracer.add("json", PromptyTracer("./traces").tracer)

@trace
def my_function():
    ...

OpenTelemetry:

from prompty.tracing.otel import otel_tracer
Tracer.add("otel", otel_tracer())