[CLEANUP][Code examples] [Example Guide][X402 Example]

Author: kyegomez

docs/examples/x402_payment_integration.md

@@ -0,0 +1,166 @@
+# X402 Payment Integration with Swarms Agents
+
+X402 is a protocol that enables seamless cryptocurrency payments for API endpoints. This guide demonstrates how to monetize your Swarms agents by integrating X402 payment requirements into your FastAPI applications.
+
+With X402, you can:
+
+| Feature                                             | Description                                  |
+|-----------------------------------------------------|----------------------------------------------|
+| Charge per API request                              | Monetize your agents on a per-call basis     |
+| Accept cryptocurrency payments                      | e.g., Base, Base Sepolia, and more           |
+| Payment gate protection for agent endpoints         | Secure endpoints with pay-to-access gates    |
+| Create pay-per-use AI services                      | Offer AI agents as on-demand paid services   |
+
+## Prerequisites
+
+Before you begin, ensure you have:
+
+- Python 3.10 or higher
+- A cryptocurrency wallet address (for receiving payments)
+- API keys for your AI model provider (e.g., OpenAI)
+- An Exa API key (if using web search functionality)
+
+## Installation
+
+Install the required dependencies:
+
+```bash
+pip install swarms x402 fastapi uvicorn python-dotenv swarms-tools
+```
+
+## Environment Setup
+
+Create a `.env` file in your project root:
+
+```bash
+# OpenAI API Key
+OPENAI_API_KEY=your_openai_api_key_here
+
+# Exa API Key (for web search)
+EXA_API_KEY=your_exa_api_key_here
+
+# Your wallet address (where you'll receive payments)
+WALLET_ADDRESS=0xYourWalletAddressHere
+```
+
+## Basic X402 Integration Example
+
+Here's a complete example of a research agent with X402 payment integration:
+
+```python
+from dotenv import load_dotenv
+from fastapi import FastAPI
+from swarms_tools import exa_search
+
+from swarms import Agent
+from x402.fastapi.middleware import require_payment
+
+# Load environment variables
+load_dotenv()
+
+app = FastAPI(title="Research Agent API")
+
+# Initialize the research agent
+research_agent = Agent(
+    agent_name="Research-Agent",
+    system_prompt="You are an expert research analyst. Conduct thorough research on the given topic and provide comprehensive, well-structured insights with citations.",
+    model_name="gpt-4o-mini",
+    max_loops=1,
+    tools=[exa_search],
+)
+
+
+# Apply x402 payment middleware to the research endpoint
+app.middleware("http")(
+    require_payment(
+        path="/research",
+        price="$0.01",
+        pay_to_address="0xYourWalletAddressHere",
+        network_id="base-sepolia",
+        description="AI-powered research agent that conducts comprehensive research on any topic",
+        input_schema={
+            "type": "object",
+            "properties": {
+                "query": {
+                    "type": "string",
+                    "description": "Research topic or question",
+                }
+            },
+            "required": ["query"],
+        },
+        output_schema={
+            "type": "object",
+            "properties": {
+                "research": {
+                    "type": "string",
+                    "description": "Comprehensive research results",
+                }
+            },
+        },
+    )
+)
+
+
+@app.get("/research")
+async def conduct_research(query: str):
+    """
+    Conduct research on a given topic using the research agent.
+
+    Args:
+        query: The research topic or question
+
+    Returns:
+        Research results from the agent
+    """
+    result = research_agent.run(query)
+    return {"research": result}
+
+
+@app.get("/")
+async def root():
+    """Health check endpoint (free, no payment required)"""
+    return {
+        "message": "Research Agent API with x402 payments",
+        "endpoints": {
+            "/research": "Paid endpoint - $0.01 per request",
+        },
+    }
+
+
+if __name__ == "__main__":
+    import uvicorn
+
+    uvicorn.run(app, host="0.0.0.0", port=8000)
+```
+
+
+## Running Your Service
+
+Start the server:
+
+```bash
+python research_agent_x402_example.py
+```
+
+Or with uvicorn directly:
+
+```bash
+uvicorn research_agent_x402_example:app --host 0.0.0.0 --port 8000 --reload
+```
+
+Your API will be available at:
+
+- Main endpoint: `http://localhost:8000/`
+
+- Research endpoint: `http://localhost:8000/research`
+
+- API docs: `http://localhost:8000/docs`
+
+
+## Next Steps
+
+1. Experiment with different pricing models
+2. Add multiple agents with specialized capabilities
+3. Implement analytics to track usage and revenue
+4. Deploy to production (see [Deployment Solutions](../deployment_solutions/overview.md))
+5. Integrate with your existing payment processing

docs/mkdocs.yml

@@ -433,6 +433,9 @@ nav:
     - AOP:
       - Medical AOP Example: "examples/aop_medical.md"
 
+    - X402:
+      - x402 Quickstart Example: "examples/x402_payment_integration.md"
+
 
   - Swarms Cloud API:
     - Overview: "swarms_cloud/migration.md"

tests/structs/test_concurrent_workflow.py

@@ -2,6 +2,7 @@
 from swarms.structs.concurrent_workflow import ConcurrentWorkflow
 import pytest
 
+
 def test_concurrent_workflow_basic_execution():
     """Test basic ConcurrentWorkflow execution with multiple agents"""
     # Create specialized agents for different perspectives
@@ -54,7 +55,9 @@ def test_concurrent_workflow_basic_execution():
     for r in result:
         assert isinstance(r, dict)
         assert "role" in r  # Agent name is stored in 'role' field
-        assert "content" in r  # Agent output is stored in 'content' field
+        assert (
+            "content" in r
+        )  # Agent output is stored in 'content' field
 
 
 def test_concurrent_workflow_with_dashboard():
@@ -106,7 +109,9 @@ def test_concurrent_workflow_with_dashboard():
     for r in result:
         assert isinstance(r, dict)
         assert "role" in r  # Agent name is stored in 'role' field
-        assert "content" in r  # Agent output is stored in 'content' field
+        assert (
+            "content" in r
+        )  # Agent output is stored in 'content' field
 
 
 def test_concurrent_workflow_batched_execution():
@@ -117,8 +122,8 @@ def test_concurrent_workflow_batched_execution():
             agent_name=f"Analysis-Agent-{i+1}",
             agent_description=f"Agent {i+1} for comprehensive business analysis",
             model_name="gpt-4o-mini",
-        verbose=False,
-        print_on=False,
+            verbose=False,
+            print_on=False,
             max_loops=1,
         )
         for i in range(4)
@@ -206,7 +211,9 @@ def test_concurrent_workflow_max_loops_configuration():
     for r in result:
         assert isinstance(r, dict)
         assert "role" in r  # Agent name is stored in 'role' field
-        assert "content" in r  # Agent output is stored in 'content' field
+        assert (
+            "content" in r
+        )  # Agent output is stored in 'content' field
 
 
 def test_concurrent_workflow_different_output_types():
@@ -318,7 +325,9 @@ def test_concurrent_workflow_real_world_scenario():
     for r in result:
         assert isinstance(r, dict)
         assert "role" in r  # Agent name is stored in 'role' field
-        assert "content" in r  # Agent output is stored in 'content' field
+        assert (
+            "content" in r
+        )  # Agent output is stored in 'content' field
 
 
 def test_concurrent_workflow_team_collaboration():
@@ -385,7 +394,10 @@ def test_concurrent_workflow_team_collaboration():
     for r in result:
         assert isinstance(r, dict)
         assert "role" in r  # Agent name is stored in 'role' field
-        assert "content" in r  # Agent output is stored in 'content' field
+        assert (
+            "content" in r
+        )  # Agent output is stored in 'content' field
+
 
 if __name__ == "__main__":
-    pytest.main([__file__, "-v"])
+    pytest.main([__file__, "-v"])

tests/structs/test_graph_workflow_comprehensive.py

@@ -222,4 +222,4 @@ def test_graph_workflow_node_metadata():
 
 
 if __name__ == "__main__":
-    pytest.main([__file__, "-v"])
+    pytest.main([__file__, "-v"])

tests/structs/test_hierarchical_swarm.py

@@ -404,4 +404,5 @@ def test_hierarchical_swarm_real_world_scenario():
 
 if __name__ == "__main__":
     import pytest
-    pytest.main([__file__, "-v"])
+
+    pytest.main([__file__, "-v"])

tests/structs/test_spreadsheet.py

@@ -21,9 +21,27 @@ def sample_csv_file(tmp_path):
     """Create a sample CSV file with agent configurations."""
     csv_path = tmp_path / "test_agents.csv"
     csv_content = [
-        ["agent_name", "description", "system_prompt", "task", "model_name"],
-        ["agent_1", "First test agent", "You are a helpful assistant. Respond with exactly 'Task completed.'", "Say hello", "gpt-4o-mini"],
-        ["agent_2", "Second test agent", "You are a code reviewer. Respond with exactly 'Review done.'", "Review this: print('hello')", "gpt-4o-mini"],
+        [
+            "agent_name",
+            "description",
+            "system_prompt",
+            "task",
+            "model_name",
+        ],
+        [
+            "agent_1",
+            "First test agent",
+            "You are a helpful assistant. Respond with exactly 'Task completed.'",
+            "Say hello",
+            "gpt-4o-mini",
+        ],
+        [
+            "agent_2",
+            "Second test agent",
+            "You are a code reviewer. Respond with exactly 'Review done.'",
+            "Review this: print('hello')",
+            "gpt-4o-mini",
+        ],
     ]
 
     with open(csv_path, "w", newline="") as f:
@@ -261,10 +279,14 @@ def test_load_from_csv_basic(sample_csv_file, temp_workspace):
     assert "agent_1" in swarm.agent_tasks
     assert "agent_2" in swarm.agent_tasks
     assert swarm.agent_tasks["agent_1"] == "Say hello"
-    assert swarm.agent_tasks["agent_2"] == "Review this: print('hello')"
+    assert (
+        swarm.agent_tasks["agent_2"] == "Review this: print('hello')"
+    )
 
 
-def test_load_from_csv_creates_agents(sample_csv_file, temp_workspace):
+def test_load_from_csv_creates_agents(
+    sample_csv_file, temp_workspace
+):
     """Test that CSV loading creates proper Agent objects."""
     agent = Agent(
         agent_name="placeholder",
@@ -349,7 +371,13 @@ def test_save_to_csv_headers(temp_workspace):
     with open(swarm.save_file_path, "r") as f:
         reader = csv.reader(f)
         headers = next(reader)
-        assert headers == ["Run ID", "Agent Name", "Task", "Result", "Timestamp"]
+        assert headers == [
+            "Run ID",
+            "Agent Name",
+            "Task",
+            "Result",
+            "Timestamp",
+        ]
 
 
 def test_save_to_csv_data(temp_workspace):