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.

47 built-in tools across 9 categories. everything is a tool — there's no special syntax for "built-in" vs "external".

tool	description	example
io
print	print data to stdout	→ print
log	log message with level	→ log level="info" "msg"
format	format data with template	→ format "hi {{.name}}"
save	save to json file	→ save path="out.json"
load	load from json file	load "data.json"
return	return data unchanged	→ return
collections
filter	filter by truthiness	list → filter
map	extract field from dicts	list → map "name"
sort	sort list	list → sort by="date"
unique	deduplicate	list → unique
flatten	flatten nested lists	[[1,2],[3]] → flatten
group	group by field	list → group by="type"
take	take first n	list → take 5
skip	skip first n	list → skip 10
count	count items	list → count
first	first item	list → first
last	last item	list → last
length	length of collection	→ length
keys	dict keys	{d} → keys
values	dict values	{d} → values
merge	merge parallel results	→ merge
json
json.parse	parse json string	'{"a":1}' → json.parse
json.stringify	serialize to json	{data} → json.stringify
http
http.get	get with auth/headers/params	"http://..." → http.get bearer="tok"
http.post	post with body	"http://..." → http.post body={}
http.put	put request	"http://..." → http.put body={}
http.patch	patch request	"http://..." → http.patch body={}
http.delete	delete request	"http://..." → http.delete
string
upper	uppercase	"hi" → upper
lower	lowercase	"HI" → lower
trim	trim whitespace	→ trim
replace	replace substring	→ replace old="a" new="b"
split	split by delimiter	→ split by=","
join	join with delimiter	→ join with=", "
contains	check substring	→ contains "needle"
math
add	add numbers	5 → add 3
sub	subtract	5 → sub 2
mul	multiply	5 → mul 3
div	divide	6 → div 2
type
type	get type name	42 → type
string	to string	42 → string
number	to number	"42" → number
system
shell	run shell command	shell "ls -la"
env	read env var	env "API_KEY"
now	current timestamp	→ now
uuid	generate uuid	→ uuid
wait	sleep seconds	wait 5
import
import	load mesh module	import "tools.mesh"
tool	define custom tool	tool name: steps: ...

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 examples/hello.mesh

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

# list all tools (47 across 9 categories)
$ python3 mesh.py --tools

# start the execution server
$ python3 meshd.py --port 8080

# run via http
$ curl -X POST http://localhost:8080/run \
    -H "Content-Type: application/json" \
    -d '{"source": "\"hello\" → upper"}'

# start the MCP server (for Cursor, Claude Code)
$ python3 mesh_mcp.py