CapMesh

1 Why CapMesh?

Every agentic application today hardcodes which tools, agents, and services it calls. When a tool changes -- or when you need to swap GitHub for GitLab, or Snyk for Semgrep -- you hunt through code, update imports, rewrite tests, and redeploy. With a dozen agents, this is a crisis every quarter.

CapMesh solves this the way DNS solved IP address management: name what you need, not where it lives. Consumers ask for a capability ("security scan"). CapMesh returns the best available provider, enforcing governance, versioning, and environment isolation along the way. The consumer calls the provider directly using the returned binding. No proxy, no overhead.

from tools.github import GitHubReader
from tools.snyk import SnykScanner
from agents.crewai import SecurityAgent

# Every dependency is hardcoded.
# Change Snyk to Semgrep?  Rewrite + redeploy.
# Swap GitHub to GitLab?   Rewrite + redeploy.
# Restrict an intern?      Add middleware.
# Audit tool usage?        Not possible.

repo    = GitHubReader(token=SECRET)
scanner = SnykScanner(api_key=OTHER_SECRET)
agent   = SecurityAgent(endpoint=HARDCODED_URL)

import capmesh

mesh = capmesh.connect()

# Consumers ask for WHAT, not WHO.
repo    = mesh.need("read a repository")
scanner = mesh.need("security vulnerability scan")
agent   = mesh.need("security code review", kind="agent")

# Swap provider?  One registry command.
# Restrict access? One YAML field.
# Audit trail?    Automatic.

import capmesh
mesh = capmesh.connect()

# Any type -- natural language
result = mesh.need("security scan")

# Only agents
agent = mesh.need("security scan", kind="agent")

# Only MCP tools
tool = mesh.need("read repo", kind="tool", protocol="mcp")

The Real Cost of Hardcoded Tools

It's not just about swapping one tool. It's what happens at scale:

What CapMesh Gives You

The DNS Analogy

Who Benefits

2 Core Concepts

Two-Plane Architecture

Four Provider Types

Architecture Diagram

3 CLI Reference -- Docker Commands Side-by-Side

CapMesh borrows Docker's mental model for the artifact plane. Every Docker workflow you know maps directly to a CapMesh command.

capmesh tool build  docker build

Validate a manifest and compute its content digest. Does not register it yet -- just proves the manifest is well-formed.

docker build -t myapp:1.0.0 .

capmesh tool build --directory ./github-reader/
capmesh agent build --directory ./security-agent/
capmesh skill build --directory ./code-review-skill/

Flags:

Example output -- building an MCP tool:

capmesh tool push  docker push

Register (push) a validated provider into the CapMesh registry. Equivalent to pushing an image to Docker Hub.

docker push myrepo/myapp:1.0.0

capmesh tool push --file ./github-reader/manifest.yaml
capmesh agent push --file ./security-agent/manifest.yaml

Flags:

capmesh tool pull  docker pull

Download a provider manifest from the registry to local disk -- useful for inspection, forking, or offline use.

docker pull myrepo/myapp:1.0.0

capmesh tool pull tools/github-reader:2.3.1
capmesh agent pull security/crewai-reviewer:3.1.0

capmesh search  docker images / docker search

List all registered providers, optionally filtering by capability, type, or status.

docker images
docker search ubuntu

capmesh search
capmesh search --capability security.code.review
capmesh search --type agent
capmesh search --query "security scan"

Flags:

capmesh tool inspect  docker inspect

Print full details of a registered provider, including governance, interface, and all capabilities.

docker inspect myrepo/myapp:1.0.0

capmesh tool inspect tools/github-reader:2.3.1
capmesh agent inspect security/crewai-reviewer:3.1.0

capmesh tag  docker tag

Create an alias pointing an existing provider to an additional name or version label.

docker tag myapp:1.0.0 myapp:latest
docker tag myapp:1.0.0 myrepo/myapp:stable

capmesh tag security/crewai-reviewer:3.1.0 \
    security/crewai-reviewer:stable

capmesh tag tools/github-reader:2.3.1 \
    tools/github-reader:latest

capmesh login  docker login

Authenticate to a CapMesh registry. Credentials are stored in ~/.capmesh/config.json.

docker login registry.example.com

capmesh login http://capmesh.acme.com:8080
capmesh login --token $CAPMESH_TOKEN http://capmesh.acme.com:8080

capmesh server start  docker run (for the registry)

Start the CapMesh registry server. Equivalent to running the registry container.

docker run -d \
  -p 5000:5000 \
  -v registry-data:/var/lib/registry \
  registry:2

capmesh server start --port 8080 \
  --db /data/capmesh.db \
  --host 0.0.0.0

Flags:

CapMesh-Only Commands (no Docker equivalent)

capmesh resolve

Resolve a capability to its best provider right now, showing the full resolution trace.

capmesh resolve <capability> [--contract v1] [--caller identity] [--env production] [--constraint ">=2.0,<3.0"] [--kind agent|tool|skill] [--protocol mcp|a2a|rest|skill] [-k] [-p] [--trace] [--json]

Examples:

capmesh resolve "security scan"
capmesh resolve "security scan" --kind agent
capmesh resolve "read a repo" --protocol mcp
capmesh resolve security.code.review -k agent -p a2a
capmesh resolve security.code.review --trace --json

Flags:

capmesh providers

List all providers for a specific capability, showing what would win resolution.

capmesh providers <capability> [--contract v1] [--verbose]

capmesh graph

Print the dependency graph for a provider -- what it requires and what provides those requirements.

capmesh graph <namespace/name:version>

4 Build Providers -- Complete Working Examples

a) Build an MCP Tool MCP

An MCP tool wraps an existing MCP server. You do NOT rewrite the server. You write a manifest.yaml that tells CapMesh what capabilities this server provides and where to find it.

Scenario: The platform team has a github-mcp MCP server running. They want to register it so any agent can discover and use it.

# ./github-reader/manifest.yaml

metadata:
  api_version: capmesh.io/v1alpha1
  kind: tool
  name: github-reader
  namespace: tools
  owner: platform-engineering
  version: "2.3.1"

governance:
  status: approved
  visibility: public
  environment: [production, staging]

interface:
  protocol: mcp
  server: github-mcp                  # the MCP server name known to the MCP client
  tools:
    - read_file
    - search_code
    - list_files

provides:
  - capability: repository.read
    contract: v1
    description: Read files and directories from a GitHub repository

  - capability: repository.search
    contract: v1
    description: Search code within a GitHub repository

Build and push:

b) Build an A2A Agent A2A

An A2A agent exposes an HTTP endpoint that implements the A2A protocol. You declare the framework label so consumers know what kind of agent to expect, but CapMesh is framework-agnostic -- it just stores the label as metadata.

Scenario: The security team has a CrewAI-based security reviewer deployed at https://crewai-security.acme.com. They register it so any pipeline can discover it.

# ./security-agent/manifest.yaml

metadata:
  api_version: capmesh.io/v1alpha1
  kind: agent
  name: crewai-security-reviewer
  namespace: security
  owner: security-engineering
  version: "3.1.0"
  labels:
    framework: crewai                # metadata only -- not enforced by CapMesh

governance:
  status: approved
  visibility: public
  environment: [production, staging]

interface:
  protocol: a2a
  endpoint: https://crewai-security.acme.com   # where the agent is ACTUALLY running

provides:
  - capability: security.code.review
    contract: v1
    description: >
      Review source code for security vulnerabilities (OWASP Top 10,
      secrets exposure, dependency risks). Returns structured findings.

requires:
  - capability: repository.read
    contract: v1
  - capability: repository.search
    contract: v1

c) Build a Skill SKILL

A Skill is a reusable procedure defined in a SKILL.md file. It contains human-readable (and LLM-readable) instructions. CapMesh resolves its tool dependencies separately at bind time, so the same skill works with GitHub, GitLab, or any other tool that satisfies the required capabilities.

Scenario: The engineering team writes a standard PR review procedure as a skill.

# ./pr-review-skill/manifest.yaml

metadata:
  api_version: capmesh.io/v1alpha1
  kind: skill
  name: pr-review
  namespace: engineering
  owner: platform-engineering
  version: "1.0.0"

governance:
  status: approved
  visibility: public
  environment: [production, staging, dev]

interface:
  protocol: skill
  skill_file: SKILL.md              # path relative to this manifest.yaml

provides:
  - capability: pr.review
    contract: v1
    description: >
      Perform a full pull request review: read changed files,
      run a security scan, check coding standards, summarize findings.

requires:
  - capability: repository.read
    contract: v1
  - capability: security.code.review
    contract: v1
  - capability: notification.send
    contract: v1

# ./pr-review-skill/SKILL.md

# PR Review Skill

## Purpose
Perform a structured pull request review using the tools provided by the caller's environment.

## Steps

1. **Read the diff**
   Use the `repository.read` tool to fetch the list of changed files in the PR.
   For each file, read its content and the diff.

2. **Security scan**
   Pass the changed file contents to the `security.code.review` tool.
   Collect findings: severity (critical/high/medium/low), file, line, description.

3. **Summarize**
   Produce a JSON report:
   ```json
   {
     "pr":       "<PR identifier>",
     "files":    "<count>",
     "findings": [ {"severity": "...", "file": "...", "line": "...", "desc": "..."} ],
     "verdict":  "approve | request_changes | needs_discussion"
   }
   ```

4. **Notify**
   Use the `notification.send` tool to post the summary to the PR thread.

## Notes
- Do NOT hardcode which tool provides repository.read or security.code.review.
  Use whatever was bound by CapMesh at resolve time.
- If security.code.review returns 0 critical findings, verdict = approve.
- If any critical finding, verdict = request_changes.

d) Build a REST Tool REST

A REST tool is any HTTP API that a consumer can call directly. CapMesh registers the base URL and any auth requirements so consumers can discover and use it without knowing the exact address.

Scenario: The security team uses SonarCloud for static analysis. They register it as a REST tool.

# ./sonarcloud-scanner/manifest.yaml

metadata:
  api_version: capmesh.io/v1alpha1
  kind: tool
  name: sonarcloud-scanner
  namespace: security
  owner: security-engineering
  version: "1.4.0"

governance:
  status: approved
  visibility: organization
  environment: [production, staging]

interface:
  protocol: rest
  endpoint: https://sonarcloud.io/api          # the actual REST API base URL
  auth:
    type: bearer
    env_var: SONARCLOUD_TOKEN                  # consumer must have this env var set

provides:
  - capability: security.static.analysis
    contract: v1
    description: Static code analysis via SonarCloud. POST /measures/component.

  - capability: security.vulnerability.report
    contract: v1
    description: Fetch vulnerability report for a component.

5 Runtime API Endpoints -- What Providers Expose

CapMesh is a registry, not a proxy. When a consumer resolves a capability, it gets back a binding -- connection information for the provider. The consumer then calls the provider directly. CapMesh is never in the data path.

The provider team runs the actual service. They just write a manifest.yaml to register it. CapMesh does not host it, proxy it, or relay traffic to it -- it simply tells consumers where to find it. This is exactly how DNS works.

What a binding looks like for each type

MCP Tool binding:

{
  "provider":  "tools/github-reader:2.3.1",
  "protocol":  "mcp",
  "connection": {
    "server":    "github-mcp",
    "tool_name": "read_file"
  },
  "trace_id":  "res_1a2b3c4d5e6f"
}

A2A Agent binding:

{
  "provider":  "security/crewai-security-reviewer:3.1.0",
  "protocol":  "a2a",
  "connection": {
    "endpoint":  "https://crewai-security.acme.com",
    "framework": "crewai"
  },
  "trace_id":  "res_8444e343826b"
}

REST Tool binding:

{
  "provider":  "security/sonarcloud-scanner:1.4.0",
  "protocol":  "rest",
  "connection": {
    "endpoint":  "https://sonarcloud.io/api",
    "auth": {
      "type":    "bearer",
      "env_var": "SONARCLOUD_TOKEN"
    }
  },
  "trace_id":  "res_9c8b7a6d5e4f"
}

Skill binding (dual binding -- instructions + tool deps resolved separately):

{
  "provider":  "engineering/pr-review:1.0.0",
  "protocol":  "skill",
  "connection": {
    "instructions": "# PR Review Skill\n\n## Steps\n...",
    "tool_bindings": [
      {
        "capability": "repository.read",
        "provider":   "tools/github-reader:2.3.1",
        "protocol":   "mcp",
        "connection": { "server": "github-mcp", "tool_name": "read_file" }
      },
      {
        "capability": "security.code.review",
        "provider":   "security/crewai-security-reviewer:3.1.0",
        "protocol":   "a2a",
        "connection": { "endpoint": "https://crewai-security.acme.com" }
      },
      {
        "capability": "notification.send",
        "provider":   "comms/slack-notifier:4.0.2",
        "protocol":   "mcp",
        "connection": { "server": "slack-mcp", "tool_name": "post_message" }
      }
    ]
  },
  "trace_id": "res_3d2c1b0a9e8f"
}

6 Use Providers -- Resolve and Call

a) Python SDK -- Resolve and Call

"""
Example: A CI pipeline that resolves a security scanner and runs it.
The pipeline does NOT know which scanner will be returned.
"""

import capmesh

# One-line setup
mesh = capmesh.connect()

# --- Resolve a capability (exact ID) ---
result = mesh.resolve("security.code.review")

print(f"Provider:  {result.provider}")       # security/crewai-security-reviewer:3.1.0
print(f"Protocol:  {result.protocol}")       # a2a
print(f"Trace ID:  {result.trace_id}")       # res_8444e343826b
print(f"Endpoint:  {result.binding.connection['endpoint']}")
# https://crewai-security.acme.com

# --- Resolve a capability (natural language) ---
binding = mesh.need("read files from a code repository")
print(f"Provider:  {binding.provider}")      # tools/github-reader:2.3.1
print(f"Protocol:  {binding.protocol}")      # mcp
print(f"Server:    {binding.connection['server']}")  # github-mcp

# --- Filter by kind and/or protocol ---
agent = mesh.need("security scan", kind="agent")
tool  = mesh.need("read repo", kind="tool", protocol="mcp")
print(f"Agent:  {agent.provider}")           # security/crewai-security-reviewer:3.1.0
print(f"Tool:   {tool.provider}")            # tools/github-reader:2.3.1

# --- Exact ID with filter ---
result = mesh.resolve("security.code.review", kind="agent")

# --- List all providers for a capability ---
providers = mesh.providers("security.code.review")
for p in providers:
    print(f"  {p.provider} ({p.protocol})")

b) Raw HTTP API -- Resolve and Return Binding

No SDK required. Any HTTP client works.

POST http://capmesh:8080/v1/resolve
Content-Type: application/json

{
  "capability":  "security.code.review",
  "contract":    "v1",
  "caller": {
    "identity":    "ci-pipeline",
    "environment": "production",
    "namespace":   "eng"
  }
}

Full response:

{
  "status":    "resolved",
  "trace_id":  "res_8444e343826b",
  "capability": "security.code.review",
  "contract":   "v1",
  "provider":   "security/crewai-security-reviewer:3.1.0",
  "protocol":   "a2a",
  "binding": {
    "connection": {
      "endpoint":  "https://crewai-security.acme.com",
      "framework": "crewai"
    }
  },
  "resolution_ms": 14,
  "selected_from": 3,
  "candidates": [
    { "provider": "security/crewai-security-reviewer:3.1.0",    "passed": true,  "reason": "selected (highest semver)" },
    { "provider": "security/langgraph-security-reviewer:2.4.0", "passed": true,  "reason": "available but lower semver" },
    { "provider": "security/code-review-skill:1.2.0",           "passed": true,  "reason": "available but lower semver" }
  ]
}

c) Using the Binding to Call the Actual Service

Once you have the binding, you call the provider directly. CapMesh is no longer involved.

import httpx
import json
from mcp import MCPClient
from a2a import A2AClient

def resolve_and_call(capability: str, payload: dict) -> dict:
    """
    Resolve a capability and call the provider.
    Works regardless of which provider is returned.
    """
    # Step 1: resolve
    resp = httpx.post("http://capmesh:8080/v1/resolve", json={
        "capability": capability,
        "contract":   "v1",
        "caller": {"identity": "my-agent", "environment": "production"}
    })
    resp.raise_for_status()
    result = resp.json()

    protocol   = result["protocol"]
    connection = result["binding"]["connection"]

    # Step 2: call the provider directly
    if protocol == "a2a":
        client = A2AClient(base_url=connection["endpoint"])
        return client.submit_task(payload)

    elif protocol == "mcp":
        client = MCPClient()
        return client.call_tool(
            server=connection["server"],
            tool=connection.get("tool_name", "default"),
            arguments=payload,
        )

    elif protocol == "rest":
        token = os.environ.get(connection["auth"]["env_var"])
        return httpx.post(
            connection["endpoint"],
            json=payload,
            headers={"Authorization": f"Bearer {token}"},
        ).json()

    elif protocol == "skill":
        instructions = connection["instructions"]
        tool_bindings = connection["tool_bindings"]
        # Execute the skill instructions using the bound tools
        return execute_skill(instructions, tool_bindings, payload)

    else:
        raise ValueError(f"Unknown protocol: {protocol}")


# Use it -- no hardcoded providers anywhere
findings = resolve_and_call("security.code.review", {
    "repo":   "acme-corp/backend-api",
    "pr":     "1247",
    "branch": "feature/auth-refactor"
})

7 Framework Integration -- Discover and Use Capabilities

Every framework follows the same 3-step pattern. The only thing that changes is how you wrap the CapMesh binding into your framework's tool type.

Step 1: Discover -- How agents find capabilities

Before using a capability, agents need to find it. Three ways:

import capmesh
mesh = capmesh.connect()

# Natural language -- agent describes what it needs
binding = mesh.need("read code from a repository")
binding = mesh.need("scan for security vulnerabilities")
binding = mesh.need("notify the team on slack")

# With kind/protocol filters
agent   = mesh.need("security scan", kind="agent")
tool    = mesh.need("read a repo", kind="tool", protocol="mcp")

# Exact ID -- when you know the capability name
binding = mesh.resolve("security.code.review")
binding = mesh.resolve("security.code.review", kind="agent")

# Explore -- see what's available (ranked by relevance)
results = mesh.discover("security")
# [
#   {capability: "security.code.review", score: 0.9, reason: "substring match"},
#   {capability: "security.scan",        score: 0.9, reason: "substring match"},
# ]

# List providers for a capability
providers = mesh.providers("security.code.review")

# Register a manifest
mesh.register("manifest.yaml")

# Search the registry
results = mesh.search("security")

Step 2: What comes back -- the Binding

Every resolution returns a binding with protocol-specific connection info:

Step 3: Use it -- Framework Examples

CrewAI

# pip install crewai capmesh
from crewai import Agent, Task, Crew
import capmesh

# Connect to CapMesh once
mesh = capmesh.connect()

# Discover capabilities (optionally filter by kind/protocol)
repo = mesh.need("read a repository", kind="tool", protocol="mcp")
scan = mesh.need("security scan", kind="agent")

# Build CrewAI tools from CapMesh bindings
def make_crewai_tool(binding):
    if binding.binding.protocol == "mcp":
        # from crewai_tools import MCPTool
        # return MCPTool(server=binding.binding.connection["server"])
        return f"MCPTool(server='{binding.binding.connection['server']}')"
    elif binding.binding.protocol == "a2a":
        # return A2ATool(endpoint=binding.binding.connection["endpoint"])
        return f"A2ATool(endpoint='{binding.binding.connection['endpoint']}')"

repo_tool = make_crewai_tool(repo)   # MCPTool resolved dynamically
scan_tool = make_crewai_tool(scan)   # A2ATool resolved dynamically

# Create CrewAI agent with dynamic tools
# agent = Agent(
#     role="Security Reviewer",
#     goal="Find vulnerabilities in code",
#     tools=[repo_tool, scan_tool],    # DYNAMIC -- not hardcoded
# )
# task = Task(description="Review PR #42", agent=agent)
# crew = Crew(agents=[agent], tasks=[task])
# result = crew.kickoff()

LangGraph / LangChain

# pip install langgraph langchain capmesh
from langchain_core.tools import tool
from langgraph.graph import StateGraph
import capmesh

mesh = capmesh.connect()

# Resolve tools at graph build time (filter by kind/protocol as needed)
repo_binding = mesh.need("read repository code", kind="tool", protocol="mcp")
scan_binding = mesh.need("security review", kind="agent")

# Create LangChain tools from bindings
@tool
def read_repo(repo: str) -> dict:
    """Read repository files. Provider resolved by CapMesh."""
    conn = repo_binding.binding.connection
    if repo_binding.binding.protocol == "mcp":
        return mcp_client.call(conn["server"], {"repo": repo})
    return httpx.get(conn["endpoint"], params={"repo": repo}).json()

@tool
def security_scan(files: list) -> dict:
    """Run security scan. Provider resolved by CapMesh."""
    conn = scan_binding.binding.connection
    return a2a_client.submit_task(conn["endpoint"], {"files": files})

# Build LangGraph
# graph = StateGraph(ReviewState)
# graph.add_node("read", read_repo)
# graph.add_node("scan", security_scan)
# graph.add_edge("read", "scan")
# app = graph.compile()
# result = app.invoke({"repo": "myorg/webapp"})

AWS Strands

# pip install strands-agents capmesh
import capmesh

mesh = capmesh.connect()

# Discover what's available
results = mesh.discover("repository")
print(f"Found {len(results)} capabilities matching 'repository':")
for r in results:
    print(f"  [{r.score:.1f}] {r.capability} -- {r.reason}")

# Resolve the best one (filter to MCP tools only)
repo = mesh.need("read repository", kind="tool", protocol="mcp")

# Build Strands agent
# from strands import Agent
# from strands.tools import MCPTool, HTTPTool
#
# if repo.binding.protocol == "mcp":
#     tool = MCPTool(repo.binding.connection["server"])
# elif repo.binding.protocol == "rest":
#     tool = HTTPTool(repo.binding.connection["endpoint"])
#
# agent = Agent(tools=[tool])
# result = agent("Read the auth module from myorg/webapp")

Microsoft AutoGen

# pip install autogen-agentchat capmesh
import capmesh

mesh = capmesh.connect()

# Resolve and create tool functions
def capmesh_tool(capability: str, kind: str = None, protocol: str = None):
    result = mesh.need(capability, kind=kind, protocol=protocol)
    conn = result.binding.connection

    def tool_fn(**kwargs):
        if result.binding.protocol == "a2a":
            return a2a_client.call(conn["endpoint"], kwargs)
        elif result.binding.protocol == "mcp":
            return mcp_client.call(conn["server"], kwargs)
        elif result.binding.protocol == "rest":
            return httpx.post(conn["endpoint"], json=kwargs).json()

    tool_fn.__name__ = capability.replace(".", "_")
    return tool_fn

# Register tools with AutoGen (filter by kind/protocol to be explicit)
read_repo = capmesh_tool("repository.read", kind="tool", protocol="mcp")
scan_code = capmesh_tool("security.code.review", kind="agent")

# from autogen import AssistantAgent, UserProxyAgent
# assistant = AssistantAgent("reviewer", llm_config={...})
# assistant.register_function({"read_repo": read_repo, "scan_code": scan_code})
# user_proxy = UserProxyAgent("user")
# user_proxy.initiate_chat(assistant, message="Review PR #42")

Google Agent Development Kit (ADK)

# pip install google-adk capmesh
import capmesh

mesh = capmesh.connect()

# Resolve multiple capabilities at once (with kind/protocol filters)
bindings = {
    "read a repo":       mesh.need("read a repo",       kind="tool", protocol="mcp"),
    "security scan":     mesh.need("security scan",     kind="agent"),
    "send notification": mesh.need("send notification", kind="tool"),
}

# Convert to Google ADK tools
# from google.adk import Agent
# from google.adk.tools import FunctionTool
#
# adk_tools = []
# for cap, binding in bindings.items():
#     conn = binding.binding.connection
#     if binding.binding.protocol == "mcp":
#         adk_tools.append(FunctionTool(name=cap,
#             fn=lambda **kw: mcp_client.call(conn["server"], kw)))
#     elif binding.binding.protocol == "a2a":
#         adk_tools.append(FunctionTool(name=cap,
#             fn=lambda **kw: a2a_client.call(conn["endpoint"], kw)))
#
# agent = Agent(model="gemini-2.0-flash", tools=adk_tools)
# result = agent.run("Review this repository for security issues")

OpenAI Agents SDK

# pip install openai capmesh
import capmesh

mesh = capmesh.connect()

# Generate OpenAI function schemas from CapMesh bindings
def make_openai_tool(capability: str, kind: str = None, protocol: str = None):
    result = mesh.need(capability, kind=kind, protocol=protocol)
    return {
        "type": "function",
        "function": {
            "name": capability.replace(".", "_").replace(" ", "_"),
            "description": f"Resolved: {result.provider_name}:{result.provider_version} via {result.binding.protocol}",
            "parameters": {"type": "object", "properties": {
                "input": {"type": "string", "description": "Input data"}
            }},
        },
        "_capmesh_binding": result.binding,  # keep binding for dispatch
    }

tools = [
    make_openai_tool("repository.read",      kind="tool", protocol="mcp"),
    make_openai_tool("security.code.review", kind="agent"),
]

# from openai import OpenAI
# client = OpenAI()
# response = client.chat.completions.create(
#     model="gpt-4",
#     messages=[{"role": "user", "content": "Review PR #42"}],
#     tools=tools,
# )
# # Dispatch tool calls to CapMesh-resolved providers
# for call in response.choices[0].message.tool_calls:
#     binding = tools[call.function.name]._capmesh_binding
#     result = dispatch(binding, call.function.arguments)

Microsoft Semantic Kernel

# pip install semantic-kernel capmesh
import capmesh

mesh = capmesh.connect()

# Resolve and create SK kernel functions (filter by kind/protocol)
repo = mesh.need("read a repository", kind="tool", protocol="mcp")
scan = mesh.need("security scan", kind="agent")

# import semantic_kernel as sk
# from semantic_kernel.functions import kernel_function
#
# kernel = sk.Kernel()
#
# @kernel_function(name="read_repo")
# def read_repo(repo_name: str) -> str:
#     conn = repo.binding.connection
#     return mcp_client.call(conn["server"], {"repo": repo_name})
#
# @kernel_function(name="security_scan")
# def security_scan(code: str) -> str:
#     conn = scan.binding.connection
#     return a2a_client.call(conn["endpoint"], {"code": code})
#
# kernel.add_function("tools", read_repo)
# kernel.add_function("tools", security_scan)
# result = await kernel.invoke("tools", "read_repo", repo_name="myorg/app")

Any Language -- Raw HTTP (Python, JavaScript, Go, Java, curl)

No SDK required. Just one HTTP call.

# Python
import httpx
resp = httpx.post("http://capmesh:8080/v1/resolve", json={
    "capability": "security.code.review", "contract": "v1",
    "caller": {"identity": "my-app"}
})
binding = resp.json()
# Use binding["binding"]["endpoint"] to call the provider

# curl
curl -X POST http://capmesh:8080/v1/resolve \
  -H "Content-Type: application/json" \
  -d '{"capability":"security.code.review","contract":"v1","caller":{"identity":"my-app"}}'

// JavaScript
const resp = await fetch("http://capmesh:8080/v1/resolve", {
  method: "POST",
  headers: {"Content-Type": "application/json"},
  body: JSON.stringify({capability: "security.code.review", contract: "v1", caller: {identity: "my-app"}})
});
const binding = await resp.json();
// Use binding.binding.endpoint

// Go
body := `{"capability":"security.code.review","contract":"v1","caller":{"identity":"my-app"}}`
resp, _ := http.Post("http://capmesh:8080/v1/resolve", "application/json", strings.NewReader(body))

Caching -- Repeated Resolutions are Fast

CapMesh caches resolution results with a 30-second TTL. Cache auto-invalidates when providers are registered, deleted, or deprecated.

import capmesh
mesh = capmesh.connect()

# First call: hits SQLite (~16ms)
mesh.need("security scan")

# Second call: cache hit (~0ms)
mesh.need("security scan")

# Register a new provider: cache auto-invalidates
mesh.register("new_scanner_manifest.yaml")

# Next call: fresh resolution (~16ms), picks new provider
mesh.need("security scan")

# Check cache performance
print(mesh.cache_stats)
# {"hits": 142, "misses": 18, "size": 12}

8 Backward Compatibility -- Zero Code Changes to Providers

Existing A2A Agent -- No Changes

You have a LangGraph agent deployed at https://my-agent.acme.com. It has been running for months. You want to make it discoverable through CapMesh.

What you do: Write one manifest.yaml. Push it. Done.

# manifest.yaml -- the ONLY thing you need
metadata:
  api_version: capmesh.io/v1alpha1
  kind: agent
  name: my-existing-langgraph-agent
  namespace: ml
  owner: ml-team
  version: "1.0.0"
  labels:
    framework: langgraph              # metadata only

governance:
  status: approved
  visibility: public
  environment: [production]

interface:
  protocol: a2a
  endpoint: https://my-agent.acme.com   # already running, no changes needed

provides:
  - capability: data.analysis
    contract: v1
    description: Analyze datasets and produce statistical summaries

Existing MCP Server -- No Changes

You have an MCP server named my-github-mcp in your MCP client config. It has been running for months. You want to make it discoverable through CapMesh.

# manifest.yaml
metadata:
  api_version: capmesh.io/v1alpha1
  kind: tool
  name: my-github-mcp-tool
  namespace: tools
  owner: platform-eng
  version: "3.0.0"

governance:
  status: approved
  visibility: public
  environment: [production, staging]

interface:
  protocol: mcp
  server: my-github-mcp             # the existing server name -- no changes needed
  tools:
    - read_file
    - list_files

provides:
  - capability: repository.read
    contract: v1
    description: Read files from a GitHub repository via MCP

No SDK Required on the Provider Side

9 In Action -- Three Scenarios

a) PR Review Pipeline

A CI pipeline resolves 4 capabilities, chains results, and produces a final report. The pipeline never hardcodes which tools to use.

"""pr_review_pipeline.py"""
import httpx, json

CAPMESH = "http://capmesh:8080"
PR = {"repo": "acme-corp/backend-api", "pr": "1247", "branch": "feature/auth-refactor"}

def resolve(capability):
    r = httpx.post(f"{CAPMESH}/v1/resolve", json={
        "capability": capability, "contract": "v1",
        "caller": {"identity": "ci-pipeline", "environment": "production"}
    })
    r.raise_for_status()
    return r.json()

# Step 1: Find the repo reader
repo_binding = resolve("repository.read")

# Step 2: Find a security reviewer
sec_binding = resolve("security.code.review")

# Step 3: Find a code quality checker
quality_binding = resolve("code.quality.check")

# Step 4: Find a notification channel
notify_binding = resolve("notification.send")

print(json.dumps({
    "repository.read":    repo_binding["provider"],
    "security.review":    sec_binding["provider"],
    "quality.check":      quality_binding["provider"],
    "notification.send":  notify_binding["provider"],
}, indent=2))

# Now call each provider using its binding...
# (abbreviated: see section 6c for full call logic)

b) Provider Swap -- Zero Code Changes

The security team decides to replace Snyk with Semgrep. The pipeline code does not change. The swap takes 30 seconds.

Before:

The security team pushes a new version of Semgrep:

$ capmesh tool push --file ./semgrep-scanner/manifest.yaml
# manifest.yaml has version: "3.0.0" -- higher than snyk:2.1.0

After: All pipelines that resolve security.static.analysis/v1 now get Semgrep 3.0.0 automatically. No pipeline code changed.

If the team needs to roll back, they deprecate Semgrep 3.0.0:

capmesh tool deprecate security/semgrep-scanner:3.0.0

c) Dynamic Discovery -- Register at Runtime, Immediately Available

A new team just deployed a specialized code reviewer for Python at https://python-reviewer.acme.com. They register it at runtime. No restarts needed anywhere.

import httpx, json

# Register via HTTP API -- no CLI needed
manifest = {
    "metadata": {
        "api_version": "capmesh.io/v1alpha1",
        "kind":        "agent",
        "name":        "python-security-reviewer",
        "namespace":   "security",
        "owner":       "python-guild",
        "version":     "1.0.0",
        "labels":      {"framework": "strands", "language": "python"}
    },
    "governance": {
        "status":      "approved",
        "visibility":  "organization",
        "environment": ["production", "staging"]
    },
    "interface": {
        "protocol": "a2a",
        "endpoint": "https://python-reviewer.acme.com"
    },
    "provides": [
        {
            "capability":  "security.code.review",
            "contract":    "v1",
            "description": "Python-specialized security review: type confusion, injection, async pitfalls"
        }
    ]
}

resp = httpx.post(
    "http://capmesh:8080/v1/providers/register",
    json=manifest,
    headers={"Authorization": "Bearer $CAPMESH_TOKEN"}
)
print(resp.json())

10 Governance -- Policies, Visibility, Approval, Audit

Approval Lifecycle

Every provider has a status in its governance block. Only approved providers are resolvable.

governance:
  status: approved      # approved | deprecated | revoked

Changing status via CLI:

Environment Isolation

governance:
  environment: [production]          # staging callers CANNOT resolve this

A staging CI pipeline resolving database.write will NEVER get the production database provider. Automatic. No extra firewall rules, no middleware.

Visibility Controls

governance:
  visibility: private           # only owner can resolve
  # visibility: organization   # only callers in same org as owner
  # visibility: public         # anyone

Version Pinning

Consumers can require a specific semver range, regardless of what version exists at the top of the registry.

import capmesh
mesh = capmesh.connect()

# Exact ID resolve with kind filter
result = mesh.resolve("security.code.review", kind="agent")

# For version constraints, use the CLI or raw HTTP API:
# capmesh resolve security.code.review --constraint ">=2.0.0,<3.0.0" --kind agent

Immutable Versions

Once an approved version is pushed, its digest is locked. You cannot push different content under the same version number.

Full Audit Trail

Every resolution is stored in SQLite with full candidate list.

GET http://capmesh:8080/v1/resolutions/res_8444e343826b

{
  "trace_id":              "res_8444e343826b",
  "timestamp":             "2026-09-23T14:32:15Z",
  "caller": {
    "identity":    "ci-pipeline",
    "environment": "production",
    "namespace":   "eng"
  },
  "requested_capability":  "security.code.review",
  "requested_contract":    "v1",
  "version_constraint":    null,
  "candidates_evaluated": [
    {
      "provider": "security/crewai-security-reviewer:3.1.0",
      "passed":   true,
      "reason":   "selected -- highest semver"
    },
    {
      "provider": "security/langgraph-security-reviewer:2.4.0",
      "passed":   true,
      "reason":   "available -- lower semver"
    },
    {
      "provider": "security/openai-reviewer:2.0.0",
      "passed":   false,
      "reason":   "governance.status=deprecated"
    }
  ],
  "selected_provider":     "security/crewai-security-reviewer:3.1.0",
  "protocol":              "a2a",
  "outcome":               "success",
  "resolution_ms":         16.0
}

Query all resolutions for a capability in the last hour:

GET http://capmesh:8080/v1/resolutions?capability=security.code.review&since=2026-09-23T13:00:00Z

11 API Reference

All endpoints are on http://<capmesh-host>:8080. Responses are JSON. The server returns standard HTTP status codes.

Registry Endpoints

POST /v1/providers/register

Request body: full manifest YAML/JSON as described in section 4.

POST /v1/providers/register
Content-Type: application/json

{ ...manifest object... }

# 201 Created
{
  "status":   "registered",
  "provider": "security/crewai-security-reviewer:3.1.0",
  "digest":   "sha256:3f9f55a4...",
  "capabilities_indexed": ["security.code.review/v1"],
  "immediately_resolvable": true
}

# 409 Conflict (version exists with different digest)
{
  "error":    "version_conflict",
  "message":  "security/crewai-security-reviewer:3.1.0 already exists with digest sha256:3f9f55a4...",
  "hint":     "Bump the version number to publish new content"
}

GET /v1/capabilities/{id}/providers

GET /v1/capabilities/security.code.review/providers?contract=v1&status=approved

{
  "capability":  "security.code.review",
  "contract":    "v1",
  "providers": [
    {
      "namespace": "security",
      "name":      "crewai-security-reviewer",
      "version":   "3.1.0",
      "protocol":  "a2a",
      "status":    "approved",
      "digest":    "sha256:3f9f55a4..."
    },
    {
      "namespace": "security",
      "name":      "langgraph-security-reviewer",
      "version":   "2.4.0",
      "protocol":  "a2a",
      "status":    "approved",
      "digest":    "sha256:9b2c71e8..."
    }
  ],
  "total": 2
}

POST /v1/search

POST /v1/search
{
  "query":   "security scan",
  "type":    "tool",
  "status":  "approved",
  "limit":   10
}

{
  "results": [
    {
      "namespace":    "security",
      "name":         "sonarcloud-scanner",
      "version":      "1.4.0",
      "kind":         "tool",
      "protocol":     "rest",
      "capabilities": ["security.static.analysis/v1", "security.vulnerability.report/v1"],
      "score":        0.91
    }
  ],
  "total": 1
}

Resolution Endpoints

POST /v1/resolve

POST /v1/resolve
{
  "capability":          "security.code.review",
  "contract":            "v1",
  "version_constraint":  ">=2.0.0",
  "caller": {
    "identity":          "ci-pipeline",
    "environment":       "production",
    "namespace":         "eng"
  }
}

# 200 OK
{
  "status":          "resolved",
  "trace_id":        "res_8444e343826b",
  "capability":      "security.code.review",
  "contract":        "v1",
  "provider":        "security/crewai-security-reviewer:3.1.0",
  "protocol":        "a2a",
  "binding": {
    "connection": {
      "endpoint":    "https://crewai-security.acme.com",
      "framework":   "crewai"
    }
  },
  "resolution_ms":   14,
  "selected_from":   3,
  "candidates": [
    { "provider": "security/crewai-security-reviewer:3.1.0",    "passed": true,  "reason": "selected" },
    { "provider": "security/langgraph-security-reviewer:2.4.0", "passed": true,  "reason": "lower semver" },
    { "provider": "security/code-review-skill:1.2.0",           "passed": true,  "reason": "lower semver" }
  ]
}

# 404 Not Found
{
  "status":  "failed",
  "error":   "no_provider",
  "message": "no approved provider satisfies security.code.review/v1 for env=dev",
  "trace_id": "res_0000000000001"
}

POST /v1/need (natural language)

POST /v1/need
{
  "query": "notify the team about a build failure",
  "caller": {
    "identity":    "ci-pipeline",
    "environment": "production"
  }
}

{
  "status":        "resolved",
  "matched_capability": "notification.send",
  "match_score":   0.88,
  "trace_id":      "res_2e3f4a5b6c7d",
  "provider":      "comms/slack-notifier:4.0.2",
  "protocol":      "mcp",
  "binding": {
    "connection": {
      "server":    "slack-mcp",
      "tool_name": "post_message"
    }
  },
  "resolution_ms": 18
}

Artifact Endpoints

Server Endpoints

GET /healthz

{"status": "ok", "version": "0.9.0", "providers": 34, "resolutions_today": 1247}

GET /v1/stats

{
  "providers": {
    "total":      34,
    "approved":   28,
    "deprecated":  4,
    "revoked":     2
  },
  "capabilities": {
    "total_ids":  19
  },
  "resolutions": {
    "total":      94321,
    "today":      1247,
    "avg_ms":     15.4,
    "success_rate": 0.997
  }
}

12 Getting Started

Install

pip install capmesh
pip install "capmesh[server]"       # for the registry server component

Start the server

capmesh server start --port 8080

Register your first provider

# 1. Write a manifest.yaml (see section 4 for full examples)
# 2. Build and push
capmesh tool build --directory ./my-tool/
capmesh tool push --file ./my-tool/manifest.yaml

# 3. Verify
capmesh tool inspect tools/my-tool:1.0.0

Resolve your first capability

capmesh resolve my.capability --caller test --env dev

Run the full demo

git clone https://github.com/capmesh/capmesh.git
cd capmesh
pip install -e ".[server,dev]"

python demo/registry/generate_providers.py   # generates 34 sample providers
python demo/app.py                           # runs the full orchestrator demo

# Step-by-step demos
python demo/01_build.py            # validate manifests, show digests
python demo/02_register.py         # register all providers
python demo/03_resolve.py          # resolve capabilities, show trace
python demo/04_benefits.py         # WITH vs WITHOUT comparison
python demo/05_production.py       # production HTTP API usage
python demo/06_governance.py       # status, visibility, version pinning
python demo/07_natural_language.py # mesh.need() demo

Run tests

pytest tests/ -v                           # 164 tests
pytest tests/ -v --cov=capmesh             # with coverage (87%)