diff --git a/ai-hub/integration_tests/test_agents.py b/ai-hub/integration_tests/test_agents.py index 1a87dd4..cb76511 100644 --- a/ai-hub/integration_tests/test_agents.py +++ b/ai-hub/integration_tests/test_agents.py @@ -110,3 +110,78 @@ # Cleanup Node client.delete(f"{BASE_URL}/nodes/admin/{node_id}", params={"admin_id": admin_id}) + +def test_agent_webhook_trigger(): + """ + Test Agent Webhook Triggering: + 1. Deploy agent with webhook trigger + 2. Obtain secret token + 3. Call webhook with custom prompt + token + 4. Verify response contains custom prompt indicator + """ + node_id = f"test-webhook-node-{uuid.uuid4().hex[:8]}" + admin_id = os.getenv("SYNC_TEST_USER_ID", "") + + with httpx.Client(timeout=10.0) as client: + # 1. Register a test node + node_payload = { + "node_id": node_id, + "display_name": "Webhook Test Node", + "is_active": True, + "skill_config": {"shell": {"enabled": True}} + } + client.post(f"{BASE_URL}/nodes/admin", params={"admin_id": admin_id}, json=node_payload) + + # 2. Deploy Agent with Webhook Trigger + deploy_payload = { + "name": "Webhook Agent", + "system_prompt": "You are a helpful assistant. Just acknowledge the received webhook prompt.", + "max_loop_iterations": 1, + "mesh_node_id": node_id, + "provider_name": "gemini", + "trigger_type": "webhook", + "default_prompt": "Standard Webhook Prompt", + "initial_prompt": None + } + r_deploy = client.post(f"{BASE_URL}/agents/deploy", json=deploy_payload, headers=_headers()) + assert r_deploy.status_code == 200, f"Deploy failed: {r_deploy.text}" + deploy_res = r_deploy.json() + + instance_id = deploy_res["instance_id"] + session_id = deploy_res["session_id"] + + # 3. Get the webhook secret + r_trig = client.get(f"{BASE_URL}/agents/{instance_id}/triggers", headers=_headers()) + assert r_trig.status_code == 200 + webhook_trigger = next(t for t in r_trig.json() if t["trigger_type"] == "webhook") + secret = webhook_trigger["webhook_secret"] + assert secret is not None + + # 4. Trigger the Webhook with a custom prompt + custom_msg = "INTER-AGENT-SIGNAL-BEEP-BOOP" + r_hook = client.post( + f"{BASE_URL}/agents/{instance_id}/webhook", + params={"token": secret}, + json={"prompt": f"Please respond exactly with: {custom_msg}"} + ) + assert r_hook.status_code == 202, f"Webhook trigger failed: {r_hook.text}" + + # 5. Wait for agent to process + print(f"\n[test] Waiting for agent to process webhook signal '{custom_msg}'...") + import time + found = False + for _ in range(30): + r_msgs = client.get(f"{BASE_URL}/sessions/{session_id}/messages", headers=_headers()) + msgs = r_msgs.json()["messages"] + # Look for assistant response containing our custom signal + if any(custom_msg in (m.get("content") or "") for m in msgs if m["sender"] == "assistant"): + found = True + break + time.sleep(2) + + assert found, "The agent did not process the custom webhook prompt correctly." + print("[test] Webhook custom prompt processed successfully!") + + # 6. Cleanup + client.delete(f"{BASE_URL}/agents/{instance_id}", headers=_headers()) + client.delete(f"{BASE_URL}/nodes/admin/{node_id}", params={"admin_id": admin_id})