Skip to main content
These are complete, copy-paste executors for the SDKs teams reach for most. Each wraps the SDK directly with createExecutor from @agentmark-ai/prompt-core; there’s no AgentMark adapter package to install or version to track. You own the code: copy the closest one, drop it in, adjust the model mapping, and you’re done. Every executor here takes the neutral render (formatted.text_config.model_name, formatted.messages, and for object prompts formatted.object_config.schema) and returns { text | object, usage }. The builder turns that into AgentMark’s wire protocol for you. If your SDK isn’t listed, the shape is identical. See Bring your own SDK for the full guide.

​
Vercel AI SDK

Wraps the ai package’s generateText / streamText (and generateObject for structured output). This example does text with both one-shot and streaming paths: 
import { createExecutor } from "@agentmark-ai/prompt-core";
import { generateText, streamText, generateObject, jsonSchema } from "ai";
import { openai } from "@ai-sdk/openai";

// Minimal model resolution — strip an optional "openai/" prefix. Swap in your
// own provider map (anthropic, google, …) keyed off the model name as needed.
const model = (name: string) => openai(name.replace(/^openai\//, ""));

export const executor = createExecutor({
  name: "vercel-ai-sdk",
  text: async (formatted, ctx) => {
    const { text, usage } = await generateText({
      model: model(formatted.text_config.model_name),
      messages: formatted.messages,
      abortSignal: ctx.signal,
    });
    // AI SDK v5 usage is already canonical ({ inputTokens, outputTokens }).
    // On v4 it's { promptTokens, completionTokens } — rename those two.
    return { text, usage: { inputTokens: usage.inputTokens, outputTokens: usage.outputTokens } };
  },
  streamText: async function* (formatted, ctx) {
    const result = streamText({
      model: model(formatted.text_config.model_name),
      messages: formatted.messages,
      abortSignal: ctx.signal,
    });
    // Consume `fullStream`, not `textStream` — a failed model call (bad API
    // key, rate limit, unknown model) surfaces ONLY as an `error` part on
    // fullStream. textStream just ends silently, and `await result.usage`
    // then rejects with the AI SDK's generic "No output generated. Check the
    // stream for errors." — hiding the real cause from the dashboard.
    for await (const part of result.fullStream) {
      if (part.type === "error") throw part.error; // builder emits it as a terminal error event
      if (part.type === "text-delta") yield { type: "text-delta", text: part.text };
      if (part.type === "finish") {
        const usage = part.totalUsage ?? part.usage;
        yield {
          type: "finish",
          reason: part.finishReason ?? "stop",
          usage: { inputTokens: usage.inputTokens, outputTokens: usage.outputTokens },
        };
      }
    }
  },
  object: async (formatted, ctx) => {
    const { object, usage } = await generateObject({
      model: model(formatted.object_config.model_name),
      messages: formatted.messages,
      schema: jsonSchema(formatted.object_config.schema),
      abortSignal: ctx.signal,
    });
    return { object, usage: { inputTokens: usage.inputTokens, outputTokens: usage.outputTokens } };
  },
});

​
OpenAI (raw SDK)

The official openai package’s chat.completions.create. Messages map straight through; structured output uses JSON-schema response format: 
import { createExecutor } from "@agentmark-ai/prompt-core";
import OpenAI from "openai";

const openai = new OpenAI();
const model = (name: string) => name.replace(/^openai\//, "");

export const executor = createExecutor({
  name: "openai",
  text: async (formatted, ctx) => {
    const res = await openai.chat.completions.create(
      { model: model(formatted.text_config.model_name), messages: formatted.messages },
      { signal: ctx.signal },
    );
    return {
      text: res.choices[0].message.content ?? "",
      finishReason: res.choices[0].finish_reason,
      usage: { inputTokens: res.usage?.prompt_tokens ?? 0, outputTokens: res.usage?.completion_tokens ?? 0 },
    };
  },
  object: async (formatted, ctx) => {
    const res = await openai.chat.completions.create(
      {
        model: model(formatted.object_config.model_name),
        messages: formatted.messages,
        response_format: {
          type: "json_schema",
          json_schema: { name: "response", schema: formatted.object_config.schema, strict: true },
        },
      },
      { signal: ctx.signal },
    );
    return {
      object: JSON.parse(res.choices[0].message.content ?? "{}"),
      usage: { inputTokens: res.usage?.prompt_tokens ?? 0, outputTokens: res.usage?.completion_tokens ?? 0 },
    };
  },
});

​
Anthropic (raw SDK)

The @anthropic-ai/sdk messages.create. Anthropic takes system as a top-level field and requires max_tokens, so split the system message out of the neutral render: 
import { createExecutor } from "@agentmark-ai/prompt-core";
import Anthropic from "@anthropic-ai/sdk";

const anthropic = new Anthropic();
const model = (name: string) => name.replace(/^anthropic\//, "");

export const executor = createExecutor({
  name: "anthropic",
  text: async (formatted, ctx) => {
    const system = formatted.messages.filter((m) => m.role === "system").map((m) => m.content).join("\n");
    const messages = formatted.messages.filter((m) => m.role !== "system");
    const res = await anthropic.messages.create(
      {
        model: model(formatted.text_config.model_name),
        max_tokens: formatted.text_config.max_tokens ?? 1024,
        system,
        messages: messages as Anthropic.MessageParam[],
      },
      { signal: ctx.signal },
    );
    const text = res.content.map((b) => (b.type === "text" ? b.text : "")).join("");
    return { text, usage: { inputTokens: res.usage.input_tokens, outputTokens: res.usage.output_tokens } };
  },
});

​
Python (AWS Bedrock)

Bedrock’s invoke_model takes a different request shape from OpenAI: anthropic_version lives in the body (not a header), max_tokens is required, and the model ID is a full cross-region inference profile ID — not the short alias you put in the prompt’s model_name field. Map it explicitly. The runner automatically stamps gen_ai.operation.name = "chat" and the config alias on the span, so the Requests view and cost attribution work without any extra code. If you want the full inference profile ID to appear in the dashboard (instead of the config alias), override gen_ai.request.model on the span after your call — set_attribute is last-write-wins: 
import json
import boto3
from agentmark.prompt_core import create_executor, ExecutorTextResult, UsageData

bedrock = boto3.client("bedrock-runtime", region_name="us-east-1")

# Map the short config alias to the full cross-region inference profile ID.
MODEL_MAP: dict[str, str] = {
    "us.anthropic.claude-opus-4-8": "us.anthropic.claude-opus-4-8-20251101-v1:0",
    "us.anthropic.claude-sonnet-4-6": "us.anthropic.claude-sonnet-4-6-20251001-v1:0",
    # add more as needed
}

def _text(formatted, ctx) -> ExecutorTextResult:
    model_alias = formatted.text_config.model_name
    model_id = MODEL_MAP.get(model_alias, model_alias)

    body = {
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens": formatted.text_config.max_tokens or 1024,
        "messages": [m.model_dump(exclude_none=True) for m in formatted.messages],
    }
    response = bedrock.invoke_model(
        modelId=model_id,
        body=json.dumps(body),
        contentType="application/json",
        accept="application/json",
    )
    payload = json.loads(response["body"].read())
    text = "".join(b["text"] for b in payload["content"] if b.get("type") == "text")
    usage = payload.get("usage", {})
    input_tokens = usage.get("input_tokens", 0)
    output_tokens = usage.get("output_tokens", 0)

    # Override the model attribute with the resolved inference profile ID so
    # the dashboard shows what actually ran, not the config alias.
    span = (ctx.extra or {}).get("span")
    if span:
        span.set_attribute("gen_ai.request.model", model_id)

    return ExecutorTextResult(
        text=text,
        usage=UsageData(input_tokens=input_tokens, output_tokens=output_tokens),
    )

executor = create_executor(name="bedrock", text=_text)

​
Python (OpenAI)

The same shape in Python: a create_executor handler returning an ExecutorTextResult with a canonical UsageData. At runtime formatted is the TextConfigSchema Pydantic model (not a dict), so use attribute access, and model_dump() the messages before handing them to your SDK: 
from agentmark.prompt_core import create_executor, ExecutorTextResult, UsageData
from openai import OpenAI

sdk = OpenAI()

def _model(name: str) -> str:
    return name.removeprefix("openai/")

def _text(formatted, ctx) -> ExecutorTextResult:
    res = sdk.chat.completions.create(
        model=_model(formatted.text_config.model_name),
        messages=[m.model_dump(exclude_none=True) for m in formatted.messages],
    )
    usage = res.usage
    return ExecutorTextResult(
        text=res.choices[0].message.content or "",
        usage=UsageData(input_tokens=usage.prompt_tokens, output_tokens=usage.completion_tokens),
    )

executor = create_executor(name="openai", text=_text)
Pass object= for structured output and stream_text= / stream_object= async generators for streaming, the same handler set as TypeScript. See Bring your own SDK for the streaming and object handlers.

​
Agent frameworks (Pydantic AI, Mastra, Claude Agent SDK)

Agent frameworks follow the identical shape; the only difference is that your handler runs an agent loop instead of a single completion. Feed the neutral render’s messages into your agent, run it, and return its final output plus token usage: 
import { createExecutor } from "@agentmark-ai/prompt-core";

export const executor = createExecutor({
  name: "my-agent-framework",
  text: async (formatted, ctx) => {
    // Build + run your agent (Pydantic AI Agent.run, a Mastra Agent, Claude's
    // query(), …) with formatted.messages, honoring ctx.signal for cancellation.
    const result = await runMyAgent(formatted.messages, { signal: ctx.signal });
    return { text: result.output, usage: { inputTokens: result.inputTokens, outputTokens: result.outputTokens } };
  },
});
For token-by-token output and tool-call/tool-result events, use a streamText handler and yield text-delta / tool-call / tool-result events as the agent emits them. See the streaming section of Bring your own SDK.

​
Serve it

Every executor above plugs into the same wiring. Register your loader and evals once, on the client (evals registered there both run in experiments and list in the New Experiment dialog), build a runner from it, and the deployed managed handler is one line: 
import { createAgentMark } from "@agentmark-ai/prompt-core";
import { createWebhookRunner } from "@agentmark-ai/sdk";

const client = createAgentMark({ loader, evals: myEvals });
const runner = createWebhookRunner({ client, executor });
export default (body) => runner.dispatch(body);
See Bring your own SDK, Path B for serving locally vs. a managed deploy.

​
Validate whatever you build

Gate any executor, copied or hand-written, with the conformance suite. One call confirms a protocol-correct stream for every kind, streaming and one-shot, including the error path: 
import { runExecutorConformance } from "@agentmark-ai/prompt-core";

await runExecutorConformance(executor, {
  text: { messages: [{ role: "user", content: "hello" }], text_config: { model_name: "openai/gpt-4o" } },
  object: { messages: [{ role: "user", content: "give me JSON" }], object_config: { model_name: "openai/gpt-4o" } },
  errorInput: { messages: null },
});
See Validate your executor for the Python equivalent and the full pattern.

​
Have questions?

Reach out any time: