pyforjs | Python 3.14+ for JavaScript engineers

Typed tool schemas from Pydantic, cosine similarity over embeddings with numpy, and an agent loop that dispatches through a tool registry, all assembled in the browser.

Everything so far pointed here. An agent is three parts, and you now have the pieces for all of them: a schema generator, a retrieval function, and a dispatch loop.

Structured output

An LLM tool call is a JSON schema plus a validator for whatever comes back. Pydantic gives you both from one class. One class covers both the schema and the validation.

tool_schema.py

Starting Python…

import json
from pydantic import BaseModel, Field
class SearchDocs(BaseModel):
    """Search the knowledge base for passages relevant to a question."""
    query: str = Field(description="Natural language search query")
    limit: int = Field(default=3, ge=1, le=10)
print(json.dumps(SearchDocs.model_json_schema(), indent=2))

That schema is what you hand to the model. When it answers, the same class parses the response, so a malformed tool call fails at the boundary with a field-level error rather than three frames deeper:

call = SearchDocs.model_validate_json(raw_arguments_from_model)
result = registry[name](call)

Vector search is one line of numpy

You do not need a vector database to retrieve over a few thousand chunks. Cosine similarity on a normalised matrix is a dot product.

cosine.py

Starting Python…

import numpy as np
# Pretend embeddings: 5 documents in 4 dimensions
rng = np.random.default_rng(0)
docs = rng.normal(size=(5, 4))
query = rng.normal(size=4)
def cosine_search(matrix, vector, k=3):
    matrix = matrix / np.linalg.norm(matrix, axis=1, keepdims=True)
    vector = vector / np.linalg.norm(vector)
    scores = matrix @ vector                      # one dot product, all rows
    top = np.argsort(-scores)[:k]
    return [(int(i), round(float(scores[i]), 3)) for i in top]
print(cosine_search(docs, query))

matrix @ vector computes every score at once in C. The keepdims=True is what makes the row-wise division broadcast correctly, dropping it is the most common bug in hand-rolled retrieval code.

The agent loop

Three steps, repeated until the model stops asking for tools:

Send the conversation plus the tool schemas.
If the reply is a tool call, look the tool up in a registry, validate the arguments, run it.
Append the result to the conversation and go back to step 1.

agent_loop.py

Starting Python…

import numpy as np
from pydantic import BaseModel, Field
# ── 1. a tool, defined once as a Pydantic model ───────────────────────────────
class Lookup(BaseModel):
    """Find the passage most relevant to a question."""
    query: str = Field(description="What to look up")
CHUNKS = [
    "Python coroutines are lazy and do nothing until awaited.",
    "asyncio.Semaphore bounds how many tasks run concurrently.",
    "The GIL prevents CPU parallelism across threads in CPython.",
]
def embed(text):
    """Stand-in for a real embedding model: a bag of characters."""
    v = np.zeros(26)
    for ch in text.lower():
        if "a" <= ch <= "z":
            v[ord(ch) - 97] += 1
    return v / (np.linalg.norm(v) or 1)
INDEX = np.array([embed(c) for c in CHUNKS])
def run_lookup(args: Lookup) -> str:
    scores = INDEX @ embed(args.query)
    return CHUNKS[int(np.argmax(scores))]
REGISTRY = {"lookup": (Lookup, run_lookup)}
# ── 2. a stubbed model: returns a tool call, then a final answer ───────────────
def fake_model(messages):
    if not any(m["role"] == "tool" for m in messages):
        return {"tool": "lookup", "arguments": '{"query": "concurrency limit"}'}
    return {"content": "Use asyncio.Semaphore — " + messages[-1]["content"]}
# ── 3. the loop ───────────────────────────────────────────────────────────────
def agent(question, max_steps=5):
    messages = [{"role": "user", "content": question}]
    for _ in range(max_steps):
        reply = fake_model(messages)
        if "content" in reply:
            return reply["content"]
        schema, fn = REGISTRY[reply["tool"]]
        args = schema.model_validate_json(reply["arguments"])   # validated at the boundary
        messages.append({"role": "tool", "content": fn(args)})
    raise RuntimeError("agent did not converge")
print(agent("How do I limit concurrency?"))

import numpy as np
from pydantic import BaseModel, Field

# ── 1. a tool, defined once as a Pydantic model ───────────────────────────────
class Lookup(BaseModel):
    """Find the passage most relevant to a question."""

query: str = Field(description="What to look up")

CHUNKS = [
    "Python coroutines are lazy and do nothing until awaited.",
    "asyncio.Semaphore bounds how many tasks run concurrently.",
    "The GIL prevents CPU parallelism across threads in CPython.",
]

def embed(text):
    """Stand-in for a real embedding model: a bag of characters."""
    v = np.zeros(26)
    for ch in text.lower():
        if "a" <= ch <= "z":
            v[ord(ch) - 97] += 1
    return v / (np.linalg.norm(v) or 1)

INDEX = np.array([embed(c) for c in CHUNKS])

def run_lookup(args: Lookup) -> str:
    scores = INDEX @ embed(args.query)
    return CHUNKS[int(np.argmax(scores))]

REGISTRY = {"lookup": (Lookup, run_lookup)}

# ── 2. a stubbed model: returns a tool call, then a final answer ───────────────
def fake_model(messages):
    if not any(m["role"] == "tool" for m in messages):
        return {"tool": "lookup", "arguments": '{"query": "concurrency limit"}'}
    return {"content": "Use asyncio.Semaphore — " + messages[-1]["content"]}

# ── 3. the loop ───────────────────────────────────────────────────────────────
def agent(question, max_steps=5):
    messages = [{"role": "user", "content": question}]

for _ in range(max_steps):
        reply = fake_model(messages)

if "content" in reply:
            return reply["content"]

schema, fn = REGISTRY[reply["tool"]]
        args = schema.model_validate_json(reply["arguments"])   # validated at the boundary
        messages.append({"role": "tool", "content": fn(args)})

raise RuntimeError("agent did not converge")

print(agent("How do I limit concurrency?"))

Notice what the loop leaves out. There is no framework, no orchestration DSL, no callback graph. A dict of tools, a validator per tool, and a bounded for. Every production agent is this plus error handling, retries and observability.

Kata

Wire the registry yourself.

Kata

Complete run_agent so it dispatches tool calls through TOOLS, validates the arguments with each tool’s Pydantic schema, appends the result to the transcript, and returns the model’s final answer. An unknown tool must raise KeyError; exceeding max_steps must raise RuntimeError.

kata_05.py

Starting Python…

import numpy as np
from pydantic import BaseModel, Field
CHUNKS = [
    "asyncio.Semaphore bounds how many coroutines run at once.",
    "Pydantic validates untrusted input at the boundary.",
    "numpy computes cosine similarity as a single matrix product.",
]
def embed(text: str) -> np.ndarray:
    v = np.zeros(26)
    for ch in text.lower():
        if "a" <= ch <= "z":
            v[ord(ch) - 97] += 1
    return v / (np.linalg.norm(v) or 1)
INDEX = np.array([embed(c) for c in CHUNKS])
class Lookup(BaseModel):
    """Retrieve the most relevant passage."""
    query: str = Field(description="What to look up")
def run_lookup(args: Lookup) -> str:
    return CHUNKS[int(np.argmax(INDEX @ embed(args.query)))]
TOOLS = {"lookup": (Lookup, run_lookup)}
def run_agent(model, question: str, max_steps: int = 5) -> str:
    """model(messages) returns {"content": str} or {"tool": str, "arguments": json}."""
    messages = [{"role": "user", "content": question}]
    # TODO: implement the loop.
    return ""

import numpy as np
from pydantic import BaseModel, Field

CHUNKS = [
    "asyncio.Semaphore bounds how many coroutines run at once.",
    "Pydantic validates untrusted input at the boundary.",
    "numpy computes cosine similarity as a single matrix product.",
]

def embed(text: str) -> np.ndarray:
    v = np.zeros(26)
    for ch in text.lower():
        if "a" <= ch <= "z":
            v[ord(ch) - 97] += 1
    return v / (np.linalg.norm(v) or 1)

INDEX = np.array([embed(c) for c in CHUNKS])

class Lookup(BaseModel):
    """Retrieve the most relevant passage."""

query: str = Field(description="What to look up")

def run_lookup(args: Lookup) -> str:
    return CHUNKS[int(np.argmax(INDEX @ embed(args.query)))]

TOOLS = {"lookup": (Lookup, run_lookup)}

def run_agent(model, question: str, max_steps: int = 5) -> str:
    """model(messages) returns {"content": str} or {"tool": str, "arguments": json}."""
    messages = [{"role": "user", "content": question}]
    # TODO: implement the loop.
    return ""

That is the whole job

A production agent adds retries, streaming, tracing and a real embedding model. The control flow does not change. If you can read the loop above you can read any agent framework's source, and usually decide you did not need it.

You are done. Every runner, kata, and quiz answer on this site ran as CPython compiled to WebAssembly, in your browser tab the whole time.

The AI engineering capstone

Structured output

Vector search is one line of numpy

The agent loop

Kata

On this page