Tutorial 10 — Plugin Marketplace¶

pip install agentmesh-marketplace                  # core marketplace package
pip install agentmesh-marketplace[cli]             # adds Click + Rich CLI commands
pip install agentmesh-platform                 # AgentMesh integration (sandbox, trust)

from pathlib import Path
from agent_marketplace import (
    PluginManifest, PluginRegistry, PluginInstaller, PluginType,
)

# 1. Create a registry (optionally backed by a JSON file)
registry = PluginRegistry(storage_path=Path(".agentmesh/registry.json"))

# 2. Register a plugin
manifest = PluginManifest(
    name="sentiment-analyzer",
    version="1.0.0",
    description="Sentiment analysis for agent responses",
    author="alice@example.com",
    plugin_type=PluginType.VALIDATOR,
    capabilities=["sentiment-scoring", "toxicity-detection"],
)
registry.register(manifest)

# 3. Search for it
results = registry.search("sentiment")
print(results[0].name)  # → "sentiment-analyzer"

# 4. Install it
installer = PluginInstaller(
    plugins_dir=Path(".agentmesh/plugins"),
    registry=registry,
)
plugin_path = installer.install("sentiment-analyzer")
print(f"Installed to {plugin_path}")

# agent-plugin.yaml
name: sentiment-analyzer
version: 1.0.0
description: Sentiment analysis for agent responses
author: alice@example.com
plugin_type: validator
capabilities:
  - sentiment-scoring
  - toxicity-detection
dependencies:
  - nlp-tokenizer>=2.0.0
  - base-validator>=1.0.0
min_agentmesh_version: "1.5.0"
signature: "LS0tLS1CRUdJTi..."          # base64-encoded Ed25519 signature

from agent_marketplace import PluginType

class PluginType(str, enum.Enum):
    POLICY_TEMPLATE = "policy_template"   # reusable governance policy templates
    INTEGRATION     = "integration"       # third-party service connectors
    AGENT           = "agent"             # autonomous agent plugins
    VALIDATOR       = "validator"         # validation / checking plugins

# Name must be non-empty, alphanumeric + hyphens/underscores
manifest = PluginManifest(name="", ...)         # ❌ MarketplaceError

# Version must be MAJOR.MINOR or MAJOR.MINOR.PATCH (all numeric)
manifest = PluginManifest(version="abc", ...)   # ❌ MarketplaceError

# Author must be non-empty
manifest = PluginManifest(author="", ...)       # ❌ MarketplaceError

from pathlib import Path
from agent_marketplace import load_manifest, save_manifest

# Load from a directory (looks for agent-plugin.yaml inside)
manifest = load_manifest(Path("./my-plugin"))

# Or load from a specific file
manifest = load_manifest(Path("./my-plugin/agent-plugin.yaml"))

# Save to a directory (writes agent-plugin.yaml)
saved_path = save_manifest(manifest, Path("./my-plugin"))
print(saved_path)  # → PosixPath("./my-plugin/agent-plugin.yaml")

from pathlib import Path
from agent_marketplace import PluginRegistry

# In-memory only (ephemeral)
registry = PluginRegistry()

# Persistent — state saved to / loaded from a JSON file
registry = PluginRegistry(storage_path=Path(".agentmesh/registry.json"))

from agent_marketplace import PluginManifest, PluginType

manifest_v1 = PluginManifest(
    name="my-policy-plugin",
    version="1.0.0",
    description="A governance policy plugin",
    author="alice@example.com",
    plugin_type=PluginType.POLICY_TEMPLATE,
    capabilities=["policy-evaluation", "constraint-checking"],
    dependencies=["base-policy>=1.0.0"],
)

registry.register(manifest_v1)

# Register a newer version of the same plugin
manifest_v2 = manifest_v1.model_copy(update={"version": "2.0.0"})
registry.register(manifest_v2)

# Duplicate name+version raises MarketplaceError
registry.register(manifest_v1)  # ❌ MarketplaceError: already registered

# Get the latest version (highest semver)
latest = registry.get_plugin("my-policy-plugin")
print(latest.version)  # → "2.0.0"

# Get a specific version
v1 = registry.get_plugin("my-policy-plugin", version="1.0.0")
print(v1.version)  # → "1.0.0"

# Non-existent plugin raises MarketplaceError
registry.get_plugin("does-not-exist")  # ❌ MarketplaceError

# Case-insensitive substring search across name and description
results = registry.search("policy")
for plugin in results:
    print(f"{plugin.name} v{plugin.version}: {plugin.description}")

# List all plugins (latest version of each)
all_plugins = registry.list_plugins()

# Filter by type
validators = registry.list_plugins(type_filter=PluginType.VALIDATOR)
integrations = registry.list_plugins(type_filter=PluginType.INTEGRATION)

# Remove a specific version
registry.unregister("my-policy-plugin", version="1.0.0")

# Remove all versions
registry.unregister("my-policy-plugin")

# Non-existent plugin raises MarketplaceError
registry.unregister("ghost-plugin")  # ❌ MarketplaceError

from cryptography.hazmat.primitives.asymmetric import ed25519
from cryptography.hazmat.primitives import serialization

# Generate a new Ed25519 private key
private_key = ed25519.Ed25519PrivateKey.generate()

# Extract the corresponding public key
public_key = private_key.public_key()

# Serialise for storage
private_bytes = private_key.private_bytes(
    encoding=serialization.Encoding.PEM,
    format=serialization.PrivateFormat.PKCS8,
    encryption_algorithm=serialization.NoEncryption(),
)
public_bytes = public_key.public_bytes(
    encoding=serialization.Encoding.PEM,
    format=serialization.PublicFormat.SubjectPublicKeyInfo,
)

# Save to files
Path("signing-key.pem").write_bytes(private_bytes)
Path("signing-key.pub").write_bytes(public_bytes)

from agent_marketplace import PluginSigner, PluginManifest, PluginType

signer = PluginSigner(private_key)

manifest = PluginManifest(
    name="sentiment-analyzer",
    version="1.0.0",
    description="Sentiment analysis for agent responses",
    author="alice@example.com",
    plugin_type=PluginType.VALIDATOR,
)

# sign() returns a new manifest with the signature field populated
signed = signer.sign(manifest)
print(signed.signature)  # → "LS0tLS1CRUdJTi..." (base64-encoded)

# The signer exposes its public key (for distributing to verifiers)
pub = signer.public_key
print(type(pub))  # → <class 'Ed25519PublicKey'>

from agent_marketplace import verify_signature

is_valid = verify_signature(signed, public_key)
print(is_valid)  # → True

from agent_marketplace import verify_signature, MarketplaceError

# Missing signature
unsigned_manifest = PluginManifest(
    name="unsigned-plugin", version="1.0.0",
    description="No sig", author="bob",
    plugin_type=PluginType.AGENT,
)
try:
    verify_signature(unsigned_manifest, public_key)
except MarketplaceError as e:
    print(e)  # signature is missing

# Tampered manifest
tampered = signed.model_copy(update={"description": "I was tampered with"})
try:
    verify_signature(tampered, public_key)
except MarketplaceError as e:
    print(e)  # signature verification failed

from cryptography.hazmat.primitives.asymmetric import ed25519
from agent_marketplace import (
    PluginSigner, PluginManifest, PluginType, verify_signature,
)

# Key pair
private_key = ed25519.Ed25519PrivateKey.generate()
public_key = private_key.public_key()

# Create and sign
manifest = PluginManifest(
    name="round-trip-demo",
    version="1.0.0",
    description="Demonstrates signing and verification",
    author="alice@example.com",
    plugin_type=PluginType.INTEGRATION,
)
signer = PluginSigner(private_key)
signed = signer.sign(manifest)

# Verify
assert verify_signature(signed, public_key) is True
print("✓ Signature verified")

from pathlib import Path
from agent_marketplace import PluginInstaller, PluginRegistry

registry = PluginRegistry(storage_path=Path(".agentmesh/registry.json"))

installer = PluginInstaller(
    plugins_dir=Path(".agentmesh/plugins"),
    registry=registry,
)

# Install the latest version
plugin_path = installer.install("sentiment-analyzer")
print(plugin_path)  # → Path(".agentmesh/plugins/sentiment-analyzer")

# Install a specific version
plugin_path = installer.install("sentiment-analyzer", version="1.0.0")

from cryptography.hazmat.primitives.asymmetric import ed25519

trusted_keys = {
    "alice@example.com": public_key,       # Ed25519PublicKey
    "bob@example.com": bobs_public_key,
}

installer = PluginInstaller(
    plugins_dir=Path(".agentmesh/plugins"),
    registry=registry,
    trusted_keys=trusted_keys,
)

# Verification happens automatically during install
plugin_path = installer.install("sentiment-analyzer", verify=True)

# Skip verification (not recommended for production)
plugin_path = installer.install("sentiment-analyzer", verify=False)

# Plugin with dependencies
manifest = PluginManifest(
    name="advanced-analyzer",
    version="1.0.0",
    description="Advanced analysis plugin",
    author="alice@example.com",
    plugin_type=PluginType.VALIDATOR,
    dependencies=[
        "nlp-tokenizer>=2.0.0",
        "base-validator>=1.0.0",
    ],
)
registry.register(manifest)

# Installing advanced-analyzer also installs nlp-tokenizer and base-validator
plugin_path = installer.install("advanced-analyzer")

installed = installer.list_installed()
for plugin in installed:
    print(f"  {plugin.name} v{plugin.version} ({plugin.plugin_type.value})")

from agent_marketplace import PluginInstaller

# The RESTRICTED_MODULES set blocks these:
#   subprocess, os, shutil, ctypes, importlib

PluginInstaller.check_sandbox("json")          # → True  (allowed)
PluginInstaller.check_sandbox("subprocess")    # → False (blocked)
PluginInstaller.check_sandbox("os")            # → False (blocked)
PluginInstaller.check_sandbox("ctypes")        # → False (blocked)

# Remove an installed plugin
installer.uninstall("sentiment-analyzer")

# Raises MarketplaceError if not installed
installer.uninstall("not-installed")  # ❌ MarketplaceError

subprocess  os  shutil  ctypes  importlib  socket  http  urllib  ftplib
smtplib  telnetlib  pickle  shelve  marshal  code  codeop  compileall
multiprocessing  signal  resource  pty  termios  fcntl  mmap  winreg  _winapi

from pathlib import Path
from agentmesh.marketplace import PluginSandbox

sandbox = PluginSandbox(
    plugins_dir=Path(".agentmesh/plugins"),
    timeout_seconds=30,                        # global default
)

result = sandbox.execute(
    plugin_name="sentiment-analyzer",
    entry_function="analyze",                  # function to call
    input_data={"text": "This product is excellent!"},
    timeout=10,                                # per-call override
)

if "result" in result:
    print(f"Score: {result['result']}")
else:
    print(f"Plugin error: {result['error']}")

from agentmesh.marketplace import PluginSandboxError

try:
    result = sandbox.execute(
        plugin_name="slow-plugin",
        entry_function="run",
        input_data={},
        timeout=5,
    )
except PluginSandboxError as e:
    print(f"Sandbox failure: {e}")

# Install latest version
agentmesh plugin install sentiment-analyzer

# Install a specific version
agentmesh plugin install sentiment-analyzer --version 1.0.0
agentmesh plugin install sentiment-analyzer -v 1.0.0

agentmesh plugin uninstall sentiment-analyzer

# List all
agentmesh plugin list

# Filter by type
agentmesh plugin list --type validator
agentmesh plugin list --type policy_template
agentmesh plugin list --type integration
agentmesh plugin list --type agent

agentmesh plugin search sentiment
agentmesh plugin search "governance policy"

# Verify by directory (looks for agent-plugin.yaml inside)
agentmesh plugin verify ./my-plugin

# Verify by manifest file
agentmesh plugin verify ./my-plugin/agent-plugin.yaml

agentmesh plugin publish ./my-plugin

from pathlib import Path
from cryptography.hazmat.primitives.asymmetric import ed25519
from cryptography.hazmat.primitives.serialization import load_pem_public_key

from agent_marketplace import (
    PluginRegistry, PluginInstaller, PluginSigner,
    PluginManifest, PluginType, verify_signature,
)

# --- Publisher side ---
private_key = ed25519.Ed25519PrivateKey.generate()

manifest = PluginManifest(
    name="trust-aware-validator",
    version="1.0.0",
    description="Validator that checks agent trust scores",
    author="governance-team@example.com",
    plugin_type=PluginType.VALIDATOR,
    capabilities=["trust-evaluation"],
    min_agentmesh_version="1.5.0",
)

signer = PluginSigner(private_key)
signed = signer.sign(manifest)

# --- Consumer side ---
registry = PluginRegistry()
registry.register(signed)

# Build a trusted-key ring from known publishers
trusted_keys = {
    "governance-team@example.com": signer.public_key,
}

installer = PluginInstaller(
    plugins_dir=Path(".agentmesh/plugins"),
    registry=registry,
    trusted_keys=trusted_keys,
)

# Install — signature is verified automatically
path = installer.install("trust-aware-validator", verify=True)
print(f"✓ Verified and installed to {path}")

response-length-validator/
├── agent-plugin.yaml
└── response_length_validator.py

# response-length-validator/agent-plugin.yaml
name: response-length-validator
version: 1.0.0
description: Rejects agent responses that exceed a configurable word limit
author: your-name@example.com
plugin_type: validator
capabilities:
  - response-length-check
  - word-counting
dependencies: []
min_agentmesh_version: "1.5.0"

# response-length-validator/response_length_validator.py
"""Response length validator plugin for the Agent Governance Toolkit."""


def validate(input_data: dict) -> dict:
    """Check whether an agent response exceeds the word limit.

    Args:
        input_data: Must contain "response" (str) and optionally
                    "max_words" (int, default 500).

    Returns:
        {"result": {"valid": bool, "word_count": int, "message": str}}
    """
    response = input_data.get("response", "")
    max_words = input_data.get("max_words", 500)

    word_count = len(response.split())
    valid = word_count <= max_words

    return {
        "result": {
            "valid": valid,
            "word_count": word_count,
            "message": (
                f"OK — {word_count} words (limit: {max_words})"
                if valid
                else f"Too long — {word_count} words exceeds limit of {max_words}"
            ),
        }
    }

from agent_marketplace import load_manifest

manifest = load_manifest("./response-length-validator")
print(manifest.name)         # → "response-length-validator"
print(manifest.plugin_type)  # → PluginType.VALIDATOR
print(manifest.capabilities) # → ["response-length-check", "word-counting"]

from cryptography.hazmat.primitives.asymmetric import ed25519
from agent_marketplace import PluginSigner, save_manifest

private_key = ed25519.Ed25519PrivateKey.generate()
signer = PluginSigner(private_key)

signed = signer.sign(manifest)
save_manifest(signed, "./response-length-validator")
print("✓ Manifest signed and saved")

from pathlib import Path
from agent_marketplace import PluginRegistry

registry = PluginRegistry(storage_path=Path(".agentmesh/registry.json"))
registry.register(signed)
print("✓ Published to registry")

agentmesh plugin publish ./response-length-validator

from pathlib import Path
from agent_marketplace import PluginInstaller
from agentmesh.marketplace import PluginSandbox

installer = PluginInstaller(
    plugins_dir=Path(".agentmesh/plugins"),
    registry=registry,
    trusted_keys={"your-name@example.com": signer.public_key},
)
installer.install("response-length-validator")

# Execute in the sandbox
sandbox = PluginSandbox(plugins_dir=Path(".agentmesh/plugins"))
result = sandbox.execute(
    plugin_name="response-length-validator",
    entry_function="validate",
    input_data={
        "response": "This is a short response.",
        "max_words": 500,
    },
)
print(result)
# → {"result": {"valid": True, "word_count": 5, "message": "OK — 5 words (limit: 500)"}}