a language for agents

mesh is a flow-based programming language designed from the ground up for ai agents. no variables, no state, no boilerplate. just pipes and tools.

try it read the docs spec

example.mesh

# fetch data, transform, output

http.get "https://api.github.com/repos/pokelabshq/council/commits"
  → json.parse
  → .[0:5]
  → for each commit:
      format "{{.sha}} — {{.commit.message}}"
  → print

features.

everything you need to write agent workflows, nothing you don't.

no state

no variables, no assignment, no mutation. data flows through pipes from left to right. what you read is what executes.

tool-native

every operation is a tool call. built-in and external tools are the same. http.get, json.parse, sentiment.analyze — all tools.

errors are data

failures don't crash the pipeline. they flow through as values. catch them with on_error:, retry with retry 3.

parallel by default

independent operations run concurrently. parallel: blocks with named branches. merge to combine results.

agent-readable

syntax is close to natural language. agents can read, write, and generate mesh without special prompting or training.

self-describing

every mesh program documents itself. the code reads like the spec. no separate documentation needed.

syntax.

here's everything mesh can do. each example is real, runnable code.

basic pipeline

# the core of mesh: pipes

http.get "https://api.example.com/data"
  → json.parse
  → .items[:5]
  → sort by: .name
  → print

data access

# access fields, index, slice

.http.get "..." → json.parse
  → .users[0]        # first
  → .name               # field
  → format "hello {{.}}"

# slicing works everywhere
.items[0:10]             # first 10
.items[-5:]               # last 5

conditionals + error handling

# if / then / else
check http.get "https://example.com/health"
  → if .status != 200:
      log "error" "down: {{.status}}"
  → else:
      log "info" "healthy"

# retry + error handling
retry 3, backoff 2s:
  http.get "https://flaky.example.com"
→ on_error:
    log "failed after 3 retries"    return {ok: false}

parallel + loops

# parallel branches
parallel:
  branch users:
    http.get "url/users" → json.parse
  branch posts:
    http.get "url/posts" → json.parse
→ merge
→ print

# for each
.items
  → for each .items:
      format "{{.name}}: {{.value}}"
      → print

tool definition

# define reusable tools
tool sentiment:
  description: "analyze text sentiment"
  input:
    text: string
  output:
    score: float
    label: string
  steps:
    http.post "http://localhost:8764/api/analyze"
      body: {text: input.text}
    → json.parse
    → format "{{.score}} ({{.label}})"

composition

# import and compose
import "./tools/social.mesh"

# real-world: daily digest
parallel:
  branch commits:
    github.commits "pokelabshq/council"
      .since:"1d"
  branch stars:
    github.stars "pokelabshq/council"
→ merge → summarize
→ telegram.send "@thealxlabs"

tools.

30+ built-in tools. everything is a tool — there's no special syntax for "built-in" vs "external".

tool	description	example
json.parse	parse json string	'str' → json.parse → .field
json.stringify	serialize to json	{data} → json.stringify
format	format as string	{name: "x"} → format "hi {{.name}}"
http.get	http get request	"http://..." → http.get → .status
http.post	http post request	"http://..." → http.post body: {}
filter	filter collection	.items → filter
map	extract field	.items → map
sort	sort collection	.items → sort by: .date
unique	deduplicate	.items → unique by: .id
flatten	flatten nested	[[1,2],[3]] → flatten
take	take first n	.items → take 5
skip	skip first n	.items → skip 10
count	count items	.items → count
first	first item	.items → first
last	last item	.items → last
length	length of collection	.items → length
keys	dict keys	{data} → keys
values	dict values	{data} → values
merge	merge parallel results	parallel: ... → merge
print	print to stdout	→ print
log	log message	→ log "info" "message"
return	return value	→ return {ok: true}
wait	wait seconds	wait 5
save	save to file	→ save "output.json"
load	load from file	load "input.json"
shell	run shell command	shell "ls -la"
env	read env var	env "API_KEY"
now	current timestamp	→ now
uuid	generate uuid	→ uuid
type	type name	42 → type
string	to string	42 → string
number	to number	"42" → number

patterns.

common patterns for real-world agent workflows. copy, modify, ship.

service monitor

health checks · alerting · retry

monitor.mesh

loop every 60s:
  check http.get "https://pokelabs.org/health"
    → timeout 10s
    → retry 2:
        http.get
    → if .status != 200:
        parallel:
          branch alert:
            telegram.send "@thealxlabs" "⚠️ pokelabs.org is down ({{.status}})"
          branch log:
            log.error "health check failed: {{.status}}"
    → else:
        log "pokelabs.org healthy ({{.response_time}}ms)"

data pipeline

fetch · transform · output

pipeline.mesh

http.get "https://api.example.com/users"
  → json.parse
  → .users
  → sort by: .created_at
  → take 20
  → for each .users:
      format "{{.name}} ({{.email}}) — {{.role}}"
  → print
  → save "report.json"

daily digest

parallel · merge · format · notify

digest.mesh

parallel:
  branch commits:
    http.get "api.github.com/repos/pokelabshq/council/commits"
      → json.parse → take 5
  branch issues:
    http.get "api.github.com/repos/pokelabshq/council/issues"
      → json.parse → count
→ merge
→ format "📊 daily: {{.commits | count}} commits, {{.issues | count}} open issues"
→ telegram.send "@thealxlabs"

why not yaml?.

mesh vs other approaches for agent workflows.

yaml workflow

name: ci
on: push
jobs:
  build:
    runs-on: ubuntu
    steps:
      - uses: actions/checkout@v4
      - run: npm test
      - name: deploy
        if: success()
        run: ./deploy.sh

mesh

check shell "npm test"
  → if .code == 0:
      shell "./deploy.sh"
      → on_error:
          log "deploy failed"
          telegram.send "@ops" "deploy failed"

feature	yaml workflows	python scripts	mesh
agent-readable	✅	❌	✅
tool-native	⚠️	❌	✅
error handling	❌	✅	✅
parallelism	❌	✅	✅
composable	❌	✅	✅
observable	❌	❌	✅
no boilerplate	✅	❌	✅
agent-writable	⚠️	❌	✅

get started.

up and running in 60 seconds.

terminal

# install
$ git clone https://github.com/pokelabshq/mesh.git
$ cd mesh

# try the repl
$ python3 mesh.py --repl
mesh> "hello" → print
hello

# run a file
$ python3 mesh.py run examples/hello.mesh

# check syntax
$ python3 mesh.py check myfile.mesh

# list all tools
$ python3 mesh.py --tools