From 1729112a0d997c73c1fc4083d6aa310c7d1ef4c9 Mon Sep 17 00:00:00 2001 From: Kye Gomez Date: Thu, 30 Oct 2025 15:33:06 -0700 Subject: [PATCH] [CLEANUP][Code examples] [Example Guide][X402 Example] --- docs/examples/x402_payment_integration.md | 166 ++++++++++++++++++ docs/mkdocs.yml | 3 + tests/structs/test_concurrent_workflow.py | 28 ++- .../test_graph_workflow_comprehensive.py | 2 +- tests/structs/test_hierarchical_swarm.py | 3 +- tests/structs/test_spreadsheet.py | 40 ++++- tests/structs/test_swarm_architectures.py | 72 +++++--- tests/utils/test_any_to_str.py | 12 +- tests/utils/test_str_to_dict.py | 70 ++++++-- 9 files changed, 336 insertions(+), 60 deletions(-) create mode 100644 docs/examples/x402_payment_integration.md diff --git a/docs/examples/x402_payment_integration.md b/docs/examples/x402_payment_integration.md new file mode 100644 index 00000000..9f6bf637 --- /dev/null +++ b/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 diff --git a/docs/mkdocs.yml b/docs/mkdocs.yml index 887cf105..44f32d80 100644 --- a/docs/mkdocs.yml +++ b/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" diff --git a/tests/structs/test_concurrent_workflow.py b/tests/structs/test_concurrent_workflow.py index 45884845..a7ba4eae 100644 --- a/tests/structs/test_concurrent_workflow.py +++ b/tests/structs/test_concurrent_workflow.py @@ -2,6 +2,7 @@ from swarms import Agent 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"]) \ No newline at end of file + pytest.main([__file__, "-v"]) diff --git a/tests/structs/test_graph_workflow_comprehensive.py b/tests/structs/test_graph_workflow_comprehensive.py index ae1c0cb7..5cb6a4a6 100644 --- a/tests/structs/test_graph_workflow_comprehensive.py +++ b/tests/structs/test_graph_workflow_comprehensive.py @@ -222,4 +222,4 @@ def test_graph_workflow_node_metadata(): if __name__ == "__main__": - pytest.main([__file__, "-v"]) \ No newline at end of file + pytest.main([__file__, "-v"]) diff --git a/tests/structs/test_hierarchical_swarm.py b/tests/structs/test_hierarchical_swarm.py index 2652cebe..565d332d 100644 --- a/tests/structs/test_hierarchical_swarm.py +++ b/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"]) \ No newline at end of file + + pytest.main([__file__, "-v"]) diff --git a/tests/structs/test_spreadsheet.py b/tests/structs/test_spreadsheet.py index d0d4ea06..639452e8 100644 --- a/tests/structs/test_spreadsheet.py +++ b/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): diff --git a/tests/structs/test_swarm_architectures.py b/tests/structs/test_swarm_architectures.py index 8f836233..cbe7d4d8 100644 --- a/tests/structs/test_swarm_architectures.py +++ b/tests/structs/test_swarm_architectures.py @@ -34,7 +34,9 @@ def create_test_agent(name: str) -> Agent: def create_test_agents(num_agents: int) -> list[Agent]: """Create specified number of test agents""" - return [create_test_agent(f"Agent{i+1}") for i in range(num_agents)] + return [ + create_test_agent(f"Agent{i+1}") for i in range(num_agents) + ] def test_circular_swarm(): @@ -65,7 +67,6 @@ def test_grid_swarm(): assert isinstance(result, list) assert len(result) > 0 - def test_linear_swarm(): @@ -77,7 +78,6 @@ def test_linear_swarm(): assert isinstance(result, list) assert len(result) > 0 - for log in result: assert "role" in log @@ -93,7 +93,6 @@ def test_star_swarm(): assert isinstance(result, list) assert len(result) > 0 - for log in result: assert "role" in log @@ -113,7 +112,6 @@ def test_mesh_swarm(): assert isinstance(result, list) assert len(result) > 0 - for log in result: assert "role" in log @@ -136,7 +134,6 @@ def test_pyramid_swarm(): assert isinstance(result, list) assert len(result) > 0 - for log in result: assert "role" in log @@ -146,67 +143,88 @@ def test_pyramid_swarm(): def test_power_swarm(): """Test power swarm mathematical pattern""" agents = create_test_agents(8) - tasks = ["Calculate in Power Swarm", "Process in Power Swarm", "Analyze in Power Swarm"] + tasks = [ + "Calculate in Power Swarm", + "Process in Power Swarm", + "Analyze in Power Swarm", + ] result = power_swarm(agents, tasks.copy()) assert isinstance(result, list) assert len(result) > 0 - def test_log_swarm(): """Test log swarm mathematical pattern""" agents = create_test_agents(8) - tasks = ["Calculate in Log Swarm", "Process in Log Swarm", "Analyze in Log Swarm"] + tasks = [ + "Calculate in Log Swarm", + "Process in Log Swarm", + "Analyze in Log Swarm", + ] result = log_swarm(agents, tasks.copy()) assert isinstance(result, list) assert len(result) > 0 - def test_exponential_swarm(): """Test exponential swarm mathematical pattern""" agents = create_test_agents(8) - tasks = ["Calculate in Exponential Swarm", "Process in Exponential Swarm", "Analyze in Exponential Swarm"] + tasks = [ + "Calculate in Exponential Swarm", + "Process in Exponential Swarm", + "Analyze in Exponential Swarm", + ] result = exponential_swarm(agents, tasks.copy()) assert isinstance(result, list) assert len(result) > 0 - def test_geometric_swarm(): """Test geometric swarm mathematical pattern""" agents = create_test_agents(8) - tasks = ["Calculate in Geometric Swarm", "Process in Geometric Swarm", "Analyze in Geometric Swarm"] + tasks = [ + "Calculate in Geometric Swarm", + "Process in Geometric Swarm", + "Analyze in Geometric Swarm", + ] result = geometric_swarm(agents, tasks.copy()) assert isinstance(result, list) assert len(result) > 0 - def test_harmonic_swarm(): """Test harmonic swarm mathematical pattern""" agents = create_test_agents(8) - tasks = ["Calculate in Harmonic Swarm", "Process in Harmonic Swarm", "Analyze in Harmonic Swarm"] + tasks = [ + "Calculate in Harmonic Swarm", + "Process in Harmonic Swarm", + "Analyze in Harmonic Swarm", + ] result = harmonic_swarm(agents, tasks.copy()) assert isinstance(result, list) assert len(result) > 0 - def test_staircase_swarm(): """Test staircase swarm pattern""" agents = create_test_agents(10) - tasks = ["Process step 1", "Process step 2", "Process step 3", "Process step 4", "Process step 5"] + tasks = [ + "Process step 1", + "Process step 2", + "Process step 3", + "Process step 4", + "Process step 5", + ] result = staircase_swarm(agents, tasks) @@ -217,7 +235,13 @@ def test_staircase_swarm(): def test_sigmoid_swarm(): """Test sigmoid swarm pattern""" agents = create_test_agents(10) - tasks = ["Sigmoid task 1", "Sigmoid task 2", "Sigmoid task 3", "Sigmoid task 4", "Sigmoid task 5"] + tasks = [ + "Sigmoid task 1", + "Sigmoid task 2", + "Sigmoid task 3", + "Sigmoid task 4", + "Sigmoid task 5", + ] result = sigmoid_swarm(agents, tasks) @@ -228,13 +252,18 @@ def test_sigmoid_swarm(): def test_sinusoidal_swarm(): """Test sinusoidal swarm pattern""" agents = create_test_agents(10) - tasks = ["Wave task 1", "Wave task 2", "Wave task 3", "Wave task 4", "Wave task 5"] + tasks = [ + "Wave task 1", + "Wave task 2", + "Wave task 3", + "Wave task 4", + "Wave task 5", + ] result = sinusoidal_swarm(agents, tasks) assert isinstance(result, list) assert len(result) > 0 - def test_one_to_one(): @@ -247,7 +276,6 @@ def test_one_to_one(): assert isinstance(result, list) assert len(result) > 0 - for log in result: assert "role" in log @@ -265,7 +293,6 @@ async def test_one_to_three(): assert isinstance(result, list) assert len(result) > 0 - @pytest.mark.asyncio @@ -279,4 +306,3 @@ async def test_broadcast(): assert isinstance(result, list) assert len(result) > 0 - diff --git a/tests/utils/test_any_to_str.py b/tests/utils/test_any_to_str.py index 1402f3e9..9e6e941e 100644 --- a/tests/utils/test_any_to_str.py +++ b/tests/utils/test_any_to_str.py @@ -57,7 +57,9 @@ class TestAnyToStr: assert "data:" in result except Exception as e: logger.error(f"Error in test_nested_dictionary: {e}") - pytest.fail(f"test_nested_dictionary failed with error: {e}") + pytest.fail( + f"test_nested_dictionary failed with error: {e}" + ) def test_tuple(self): """Test converting a tuple to string.""" @@ -99,7 +101,9 @@ class TestAnyToStr: assert result == '"hello"' except Exception as e: logger.error(f"Error in test_string_with_quotes: {e}") - pytest.fail(f"test_string_with_quotes failed with error: {e}") + pytest.fail( + f"test_string_with_quotes failed with error: {e}" + ) def test_integer(self): """Test converting an integer.""" @@ -121,4 +125,6 @@ class TestAnyToStr: assert "None" in result except Exception as e: logger.error(f"Error in test_mixed_types_in_list: {e}") - pytest.fail(f"test_mixed_types_in_list failed with error: {e}") + pytest.fail( + f"test_mixed_types_in_list failed with error: {e}" + ) diff --git a/tests/utils/test_str_to_dict.py b/tests/utils/test_str_to_dict.py index 1464daf5..329c4f51 100644 --- a/tests/utils/test_str_to_dict.py +++ b/tests/utils/test_str_to_dict.py @@ -16,7 +16,9 @@ class TestStrToDict: assert result == {"key": "value"} except Exception as e: logger.error(f"Error in test_valid_json_string: {e}") - pytest.fail(f"test_valid_json_string failed with error: {e}") + pytest.fail( + f"test_valid_json_string failed with error: {e}" + ) def test_nested_json_string(self): """Test converting a nested JSON string.""" @@ -26,7 +28,9 @@ class TestStrToDict: assert result == {"a": 1, "b": {"c": 2}} except Exception as e: logger.error(f"Error in test_nested_json_string: {e}") - pytest.fail(f"test_nested_json_string failed with error: {e}") + pytest.fail( + f"test_nested_json_string failed with error: {e}" + ) def test_list_in_json_string(self): """Test converting JSON string containing a list.""" @@ -36,7 +40,9 @@ class TestStrToDict: assert result == {"items": [1, 2, 3]} except Exception as e: logger.error(f"Error in test_list_in_json_string: {e}") - pytest.fail(f"test_list_in_json_string failed with error: {e}") + pytest.fail( + f"test_list_in_json_string failed with error: {e}" + ) def test_empty_json_object(self): """Test converting an empty JSON object.""" @@ -46,27 +52,41 @@ class TestStrToDict: assert result == {} except Exception as e: logger.error(f"Error in test_empty_json_object: {e}") - pytest.fail(f"test_empty_json_object failed with error: {e}") + pytest.fail( + f"test_empty_json_object failed with error: {e}" + ) def test_json_with_numbers(self): """Test converting JSON string with various number types.""" try: - result = str_to_dict('{"int": 42, "float": 3.14, "negative": -5}') + result = str_to_dict( + '{"int": 42, "float": 3.14, "negative": -5}' + ) assert result is not None, "Result should not be None" - assert result == {"int": 42, "float": 3.14, "negative": -5} + assert result == { + "int": 42, + "float": 3.14, + "negative": -5, + } except Exception as e: logger.error(f"Error in test_json_with_numbers: {e}") - pytest.fail(f"test_json_with_numbers failed with error: {e}") + pytest.fail( + f"test_json_with_numbers failed with error: {e}" + ) def test_json_with_booleans(self): """Test converting JSON string with boolean values.""" try: - result = str_to_dict('{"true_val": true, "false_val": false}') + result = str_to_dict( + '{"true_val": true, "false_val": false}' + ) assert result is not None, "Result should not be None" assert result == {"true_val": True, "false_val": False} except Exception as e: logger.error(f"Error in test_json_with_booleans: {e}") - pytest.fail(f"test_json_with_booleans failed with error: {e}") + pytest.fail( + f"test_json_with_booleans failed with error: {e}" + ) def test_json_with_null(self): """Test converting JSON string with null value.""" @@ -84,13 +104,17 @@ class TestStrToDict: with pytest.raises(json.JSONDecodeError): str_to_dict('{"invalid": json}') # Invalid JSON except Exception as e: - logger.error(f"Error in test_invalid_json_raises_error: {e}") - pytest.fail(f"test_invalid_json_raises_error failed with error: {e}") + logger.error( + f"Error in test_invalid_json_raises_error: {e}" + ) + pytest.fail( + f"test_invalid_json_raises_error failed with error: {e}" + ) def test_complex_nested_structure(self): """Test converting a complex nested JSON structure.""" try: - json_str = ''' + json_str = """ { "user": { "name": "John", @@ -100,7 +124,7 @@ class TestStrToDict: "tags": ["python", "testing"], "metadata": null } - ''' + """ result = str_to_dict(json_str) assert result is not None, "Result should not be None" assert result["user"]["name"] == "John" @@ -108,8 +132,12 @@ class TestStrToDict: assert result["tags"] == ["python", "testing"] assert result["metadata"] is None except Exception as e: - logger.error(f"Error in test_complex_nested_structure: {e}") - pytest.fail(f"test_complex_nested_structure failed with error: {e}") + logger.error( + f"Error in test_complex_nested_structure: {e}" + ) + pytest.fail( + f"test_complex_nested_structure failed with error: {e}" + ) def test_retries_parameter(self): """Test that retries parameter works correctly.""" @@ -120,7 +148,9 @@ class TestStrToDict: assert result == {"test": 1} except Exception as e: logger.error(f"Error in test_retries_parameter: {e}") - pytest.fail(f"test_retries_parameter failed with error: {e}") + pytest.fail( + f"test_retries_parameter failed with error: {e}" + ) def test_json_with_unicode_characters(self): """Test converting JSON string with unicode characters.""" @@ -130,5 +160,9 @@ class TestStrToDict: assert result["emoji"] == "🐍" assert result["text"] == "你好" except Exception as e: - logger.error(f"Error in test_json_with_unicode_characters: {e}") - pytest.fail(f"test_json_with_unicode_characters failed with error: {e}") + logger.error( + f"Error in test_json_with_unicode_characters: {e}" + ) + pytest.fail( + f"test_json_with_unicode_characters failed with error: {e}" + )