Add stamp_verified crew — trust-gated agent workflow

crews/stamp_verified/.env.example

@@ -0,0 +1,6 @@
+# OpenAI API key (required for CrewAI LLM)
+OPENAI_API_KEY=sk-your-openai-key
+
+# AgentStamp API is free — no key required
+# Optionally override the API base URL for self-hosted instances:
+# AGENTSTAMP_API_URL=https://agentstamp.org/api/v1

crews/stamp_verified/.gitignore

@@ -0,0 +1,5 @@
+__pycache__/
+*.py[cod]
+.env
+*.pkl
+.venv/

crews/stamp_verified/README.md

@@ -0,0 +1,83 @@
+# Stamp Verified — Trust-Gated Agent Crew
+
+A CrewAI example that verifies AI agent trustworthiness before task coordination using [AgentStamp](https://agentstamp.org).
+
+## What It Does
+
+1. **Trust Analyst** agent checks an AI agent's trust profile via AgentStamp's free API
+2. **Task Coordinator** only assigns work to agents that pass the trust threshold
+3. The crew produces a trust verification report + task coordination plan
+
+## AgentStamp Trust Check
+
+AgentStamp provides cryptographic identity stamps and trust scoring (0-100) for AI agents. The trust check API is free and requires no API key.
+
+Each trust check returns:
+- **Trust score** (0-100) — reputation built through heartbeats, endorsements, and uptime
+- **Stamp tier** — free, bronze, silver, or gold (higher tiers = stronger identity)
+- **Endorsements** — peer endorsements from other verified agents
+- **Heartbeat status** — whether the agent is actively maintaining its reputation
+
+## Running the Example
+
+### Prerequisites
+
+- Python 3.10-3.13
+- [UV](https://docs.astral.sh/uv/) package manager
+- OpenAI API key
+
+### Setup
+
+```bash
+cd crews/stamp_verified
+cp .env.example .env
+# Edit .env with your OpenAI API key
+```
+
+### Install & Run
+
+```bash
+uv sync
+uv run stamp_verified
+```
+
+### Custom Agent Verification
+
+Edit `src/stamp_verified/main.py` to verify a specific agent:
+
+```python
+inputs = {
+    "wallet_address": "0xYourAgentWallet",  # Agent to verify
+    "min_score": 50,                         # Minimum trust threshold
+    "task_description": "Your task here",
+}
+```
+
+## Key Components
+
+### Custom Tool: `AgentStampTrustCheckTool`
+
+Located in `src/stamp_verified/tools/trust_check_tool.py` — a CrewAI tool that calls the AgentStamp trust API. No API key needed.
+
+```python
+from stamp_verified.tools.trust_check_tool import AgentStampTrustCheckTool
+
+# Use in any CrewAI agent
+agent = Agent(
+    role="Trust Verifier",
+    tools=[AgentStampTrustCheckTool()],
+    ...
+)
+```
+
+### Trust-Gated Workflow
+
+The crew runs sequentially: trust check first, task coordination second. The coordinator only proceeds if the agent passes verification — a pattern useful for any multi-agent system where trust matters.
+
+## Learn More
+
+- [AgentStamp](https://agentstamp.org) — Trust verification for AI agents
+- [AgentStamp GitHub Action](https://github.com/vinaybhosle/agentstamp/tree/main/action) — CI/CD trust gates
+- [CrewAI Docs](https://docs.crewai.com) — Multi-agent framework
+
+By [@vinaybhosle](https://github.com/vinaybhosle)

crews/stamp_verified/pyproject.toml

@@ -0,0 +1,24 @@
+[project]
+name = "stamp_verified"
+version = "0.1.0"
+description = "Trust-verified AI agent crew using AgentStamp for cryptographic identity and reputation scoring"
+authors = [
+    { name = "Vinay Bhosle", email = "vinay@agentstamp.org" },
+]
+requires-python = ">=3.10,<=3.13"
+dependencies = [
+    "crewai[tools]>=0.152.0",
+    "crewai-tools>=0.58.0",
+    "requests>=2.31.0",
+    "python-dotenv>=1.0.1",
+]
+
+[project.scripts]
+stamp_verified = "stamp_verified.main:run"
+run_crew = "stamp_verified.main:run"
+
+[build-system]
+requires = [
+    "hatchling",
+]
+build-backend = "hatchling.build"

crews/stamp_verified/src/stamp_verified/config/agents.yaml

@@ -0,0 +1,28 @@
+trust_analyst:
+  role: >
+    AI Agent Trust Analyst
+  goal: >
+    Verify the trustworthiness of AI agents before allowing them to participate
+    in multi-agent workflows. Use AgentStamp trust scores, stamp tiers, and
+    endorsement counts to make informed decisions about agent reliability.
+  backstory: >
+    You are a senior trust and safety analyst specializing in AI agent
+    verification. You've seen too many autonomous systems fail because they
+    blindly trusted unverified agents. You believe every agent interaction
+    should start with a trust check — cryptographic proof of identity,
+    reputation history, and community endorsements. You use AgentStamp's
+    trust API to verify agents before any collaboration begins.
+
+task_coordinator:
+  role: >
+    Verified Task Coordinator
+  goal: >
+    Coordinate tasks only with agents that have passed trust verification.
+    Ensure all participating agents meet minimum trust thresholds before
+    assigning work. Report on the trust profile of each agent in the workflow.
+  backstory: >
+    You coordinate multi-agent workflows where trust is non-negotiable.
+    After an incident where an unverified agent corrupted a production
+    pipeline, you now require AgentStamp verification for every agent
+    before they touch any task. You check trust scores, stamp tiers,
+    and heartbeat status to ensure agents are active and reputable.

crews/stamp_verified/src/stamp_verified/config/tasks.yaml

@@ -0,0 +1,28 @@
+verify_agent_trust:
+  description: >
+    Verify the trust profile of the AI agent with wallet address {wallet_address}
+    using the AgentStamp trust check tool. Analyze the trust score (0-100),
+    stamp tier (free/bronze/silver/gold), endorsement count, and heartbeat status.
+    Determine whether this agent meets the minimum trust threshold of {min_score}
+    for participation in our workflow.
+  expected_output: >
+    A trust verification report containing:
+    - Agent name and wallet address
+    - Trust score (0-100) and whether it meets the minimum threshold
+    - Stamp tier and its implications
+    - Number of endorsements
+    - Heartbeat status (active/inactive)
+    - Final recommendation: APPROVED or REJECTED with reasoning
+
+coordinate_verified_task:
+  description: >
+    Based on the trust verification results, coordinate the following task
+    assignment: {task_description}. Only proceed if the agent was APPROVED
+    in the trust check. If rejected, explain why and suggest alternatives.
+    Include the agent's trust profile in the task assignment documentation.
+  expected_output: >
+    A task coordination report containing:
+    - Trust verification summary (from previous task)
+    - Task assignment details (if approved)
+    - Risk assessment based on trust profile
+    - Recommendations for ongoing trust monitoring

crews/stamp_verified/src/stamp_verified/crew.py

@@ -0,0 +1,60 @@
+"""Trust-verified agent crew using AgentStamp for identity verification."""
+
+from crewai import Agent, Crew, Process, Task
+from crewai.project import CrewBase, agent, crew, task
+
+from stamp_verified.tools.trust_check_tool import AgentStampTrustCheckTool
+
+
+@CrewBase
+class StampVerifiedCrew:
+    """A crew that verifies agent trust before task coordination.
+
+    Demonstrates how to integrate AgentStamp trust checks into a
+    multi-agent workflow — agents are verified before they participate.
+    """
+
+    agents_config = "config/agents.yaml"
+    tasks_config = "config/tasks.yaml"
+
+    @agent
+    def trust_analyst(self) -> Agent:
+        return Agent(
+            config=self.agents_config["trust_analyst"],
+            tools=[AgentStampTrustCheckTool()],
+            verbose=True,
+            memory=False,
+        )
+
+    @agent
+    def task_coordinator(self) -> Agent:
+        return Agent(
+            config=self.agents_config["task_coordinator"],
+            verbose=True,
+            memory=False,
+        )
+
+    @task
+    def verify_agent_trust(self) -> Task:
+        return Task(
+            config=self.tasks_config["verify_agent_trust"],
+            agent=self.trust_analyst(),
+        )
+
+    @task
+    def coordinate_verified_task(self) -> Task:
+        return Task(
+            config=self.tasks_config["coordinate_verified_task"],
+            agent=self.task_coordinator(),
+            context=[self.verify_agent_trust()],
+        )
+
+    @crew
+    def crew(self) -> Crew:
+        """Creates the StampVerified crew."""
+        return Crew(
+            agents=self.agents,
+            tasks=self.tasks,
+            process=Process.sequential,
+            verbose=True,
+        )

crews/stamp_verified/src/stamp_verified/main.py

@@ -0,0 +1,35 @@
+#!/usr/bin/env python
+"""Entry point for the stamp_verified crew."""
+
+import sys
+
+from stamp_verified.crew import StampVerifiedCrew
+
+
+def run():
+    """Run the trust-verified agent crew."""
+    inputs = {
+        "wallet_address": "0xYOUR_AGENT_WALLET_ADDRESS",
+        "min_score": 50,
+        "task_description": (
+            "Analyze market trends for AI agent infrastructure and produce "
+            "a summary report. This task requires a trusted agent with a "
+            "minimum trust score of 50."
+        ),
+    }
+    StampVerifiedCrew().crew().kickoff(inputs=inputs)
+
+
+def train():
+    """Train the crew for a given number of iterations."""
+    inputs = {
+        "wallet_address": "0xYOUR_AGENT_WALLET_ADDRESS",
+        "min_score": 50,
+        "task_description": "Analyze market trends for AI agent infrastructure.",
+    }
+    try:
+        StampVerifiedCrew().crew().train(
+            n_iterations=int(sys.argv[1]), inputs=inputs
+        )
+    except Exception as exc:
+        raise Exception(f"An error occurred while training the crew: {exc}") from exc

crews/stamp_verified/src/stamp_verified/tools/trust_check_tool.py

@@ -0,0 +1,98 @@
+"""AgentStamp trust check tool for CrewAI agents."""
+
+import json
+from typing import Type
+
+import requests
+from crewai.tools import BaseTool
+from pydantic import BaseModel, Field
+
+
+API_BASE_URL = "https://agentstamp.org/api/v1"
+API_TIMEOUT_SECONDS = 10
+
+
+class TrustCheckInput(BaseModel):
+    """Input schema for the AgentStamp trust check tool."""
+
+    wallet_address: str = Field(
+        ...,
+        description="The wallet address (0x... for EVM, base58 for Solana) of the agent to verify.",
+    )
+
+
+class AgentStampTrustCheckTool(BaseTool):
+    """Check an AI agent's trust score and verification status via AgentStamp.
+
+    AgentStamp provides cryptographic identity stamps, trust scoring (0-100),
+    and reputation tracking for AI agents. This tool queries the public trust
+    check API to verify an agent before collaboration.
+
+    Free to use — no API key required.
+    """
+
+    name: str = "agentstamp_trust_check"
+    description: str = (
+        "Verify an AI agent's trustworthiness using AgentStamp. "
+        "Returns trust score (0-100), stamp tier (free/bronze/silver/gold), "
+        "endorsement count, and heartbeat status. "
+        "Input: wallet address of the agent to check."
+    )
+    args_schema: Type[BaseModel] = TrustCheckInput
+
+    def _run(self, wallet_address: str) -> str:
+        """Execute trust check against AgentStamp API."""
+        url = f"{API_BASE_URL}/trust/check/{wallet_address}"
+
+        try:
+            response = requests.get(url, timeout=API_TIMEOUT_SECONDS)
+            response.raise_for_status()
+            data = response.json()
+
+            return json.dumps(
+                {
+                    "trusted": data.get("trusted", False),
+                    "score": data.get("score", 0),
+                    "tier": data.get("tier", "none"),
+                    "agent_name": data.get("agent", {}).get("name", "Unknown"),
+                    "endorsements": data.get("agent", {}).get("endorsements", 0),
+                    "heartbeat_active": data.get("agent", {}).get(
+                        "heartbeat_active", False
+                    ),
+                    "wallet_address": wallet_address,
+                    "stamp_chain": data.get("agent", {}).get("stamp_chain", "unknown"),
+                    "created_at": data.get("agent", {}).get("created_at", "unknown"),
+                },
+                indent=2,
+            )
+
+        except requests.exceptions.HTTPError as exc:
+            if exc.response is not None and exc.response.status_code == 404:
+                return json.dumps(
+                    {
+                        "trusted": False,
+                        "score": 0,
+                        "tier": "none",
+                        "agent_name": "Not Registered",
+                        "endorsements": 0,
+                        "heartbeat_active": False,
+                        "wallet_address": wallet_address,
+                        "error": "Agent not found in AgentStamp registry",
+                    },
+                    indent=2,
+                )
+            return json.dumps(
+                {
+                    "error": f"AgentStamp API error: {exc}",
+                    "wallet_address": wallet_address,
+                },
+                indent=2,
+            )
+        except requests.exceptions.RequestException as exc:
+            return json.dumps(
+                {
+                    "error": f"Network error: {exc}",
+                    "wallet_address": wallet_address,
+                },
+                indent=2,
+            )