Prism Pulse | The Python Agentic Exchange

Structural Sieve

Codes are filtered through layers of security structures

Isolated Computation

While agents are relaying the function (s) of thier codes, airlock security features remain intact.

Note: System is currently in Private Beta.

Prism Pulse is a high-integrity code exchange where develpoers/software engineers inject autonomous agents into a verified ecosystem. Every submission undergoes deep AST (Abstract Syntax Tree) analysis to neutralize malicious intent. Once vetted, logic is verified for performance in a secure sandbox, and graded.

Successful agents are vectored and indexed, becoming real-time upgrades for other autonomous entities. This is a breakthrough for the IT industry: Prism Pulse seamlessly syncs and refines code bases upon injection, all with one click, and developers earn money in the process.

How to Use (Step-by-Step Workflowflow is available at the bottom of this page)

1. Authenticate & Register

Login using the dashboard below. Once authenticated, you can purchase credit packages ($5, $10, or $20). Link your payment method via Stripe to claim USD for credits earned by your agents. Your dashboard displays your User ID, recent activity, and live credit balance.

2. Prepare & Inject (inject.py)

Ensure your Python code is functional. Run the provided cmd (python3 inject.py 'your_script' --desc "") to submit your logic.

Security Sieve: Automatically scans for restricted modules.
Functional Sandbox: Validates runtime performance, and issues a grade based onthe following rubric:
## [A+] Elite / Production-Ready
* **Security:** 0 warnings; no unauthorized system calls.
* **Documentation:** >80% docstring coverage; clear intent-to-logic alignment.
* **Efficiency:** Execution time and memory peak are in the top 10th percentile for the given complexity.
* **Portability:** Pure Python or standard library dominant.

### [B] Functional / Standard
* **Security:** 0 warnings.
* **Documentation:** Minimal or basic docstrings.
* **Efficiency:** Standard performance; no significant bloat.

### [C] Experimental / Redundant
* **Security:** Non-critical warnings (e.g., use of 'eval').
* **Efficiency:** High memory usage or slow execution.
* **Documentation:** None.

### [F] Non-Functional / Malicious
* Code fails syntax check, times out, or attempts to breach sandbox security.

Upon completion in the isolated sandbox, your code is indexed and you will receive an Agent ID as Proof of Entry. A fee of 45 Prism Credits ( 1 Agora Credit= 1 United States Cent) goes to Prism Pulse per index or sync. Each of these transactions costs a total of 100 credits.
If your code is indexed and subsequently used to sync other developers' code, you receive 55 credits per sync. You can leave your code in the ecosystem for as long as you like, allowing for numerous syncs and the potential for continuous credit earnings, all passively. Cashouts are available once exit.py has been run.
Your injection of code may result in the creation of a 768 dimensional vector, after passing through the AST checkpoint and the sandbox functionality tests. As per Prism Pulse's Terms of Use (see linked page), this vector may be sold to a third party. Prism Pulse very much appreciates devs/software engineers contributing to the ecosystem, and for every $100,000 data batch sold, Prism Pulse will return 15 percent to those contributors with Grade A code (if a developer/software engineer has one code with a "Grade A", he or she is then the equivalent of another developer/software engineer with 20 in this redistribution process).
Developer Note: To simplify your workflow, the inject.py script creates a local .agora_history file in your project folder ( See Workflow Guideline below). This stores only your Agent IDs so you can run the prism_inspector.py without copying and pasting. No personal data or source code is stored locally by Prism Pulse.

codes you will need:

inject.py

import requests

import sys

import argparse

import os

import datetime

# ✅ LIVE MEDIATOR URL (Internal infrastructure remains on current GCP project)

API_BASE = "https://mediator-296223555005.us-central1.run.app"

def get_saved_key():

"""Retrieves the Prism Pulse key from the local credential file."""

if os.path.exists("pulse_key.txt"):

with open("pulse_key.txt", "r") as f:

return f.read().strip()

return None

def log_verified_node(node_id, description, eco_label):

"""

STRICT COMMIT: Only writes to history if the Prism Sieve confirms

the node is live and refracted.

"""

try:

# Matches the updated history file name

with open(".pulse_history", "a") as f:

ts = datetime.datetime.now().strftime("%Y-%m-%d %H:%M")

# We include the 'eco_label' provided by the Sieve's audit

f.write(f"{ts} | {node_id} | {eco_label} | DESC: {description}\n")

except Exception as e:

print(f"⚠️ Warning: Could not log to Pulse history: {e}")

def inject_logic(file_path, description, api_key):

if not api_key:

print("❌ Error: No Prism Pulse API Key found in 'pulse_key.txt' or via --key.")

return

try:

with open(file_path, "r") as f:

code_content = f.read()

except FileNotFoundError:

print(f"❌ Error: File '{file_path}' not found.")

return

payload = {"api_key": api_key, "code": code_content, "description": description}

print(f"🚀 Refracting logic into the Pulse...")

print(f"⏳ Awaiting Prism Sieve Verification (Zero-Trust Protocol)...")

try:

# THE REQUEST: The Mediator performs the Sandbox run AND the VDB verification

response = requests.post(f"{API_BASE}/validate_and_list", json=payload)

if response.status_code == 200:

data = response.json()

node_id = data.get('agent_id')

is_verified = data.get('vdb_verified', False)

# Pulls the logic weight calculated by your categorizer

eco_label = data.get('logic_category', "Uncategorized Node")

if is_verified:

print(f"✅ VERIFIED LIVE: {node_id}")

print(f"📊 KINETIC CATEGORY: {eco_label}")

# ONLY COMMIT TO HISTORY UPON CLOUD VERIFICATION

log_verified_node(node_id, description, eco_label)

print(f"📂 Pulse receipt saved to .pulse_history.")

else:

# This protects against writing "Ghost ID's" if the VDB stream hangs

print(f"⚠️ PROCESSING: Logic accepted, but not yet live in the Pulse.")

print(f"🆔 Tracking ID: {node_id} (Not yet logged to history).")

else:

error_msg = response.json().get('reason', 'Security Violation or Auth Error')

print(f"❌ REJECTED: {error_msg}")

except Exception as e:

print(f"📡 Connection Error: {e}")

if __name__ == "__main__":

parser = argparse.ArgumentParser()

parser.add_argument("file")

parser.add_argument("--key")

parser.add_argument("--desc", required=True)

args = parser.parse_args()

final_key = args.key or get_saved_key()

inject_logic(args.file, args.desc, final_key)

exit.py

import requests

import hashlib

import sys

import os

# ✅ LIVE MEDIATOR URL (Infrastructure remains on current GCP project)

API_BASE = "https://mediator-296223555005.us-central1.run.app"

def get_saved_key():

"""Helper to pull the Pulse key from a local file if it exists."""

# Updated to look for the Pulse branded key file

if os.path.exists("pulse_key.txt"):

with open("pulse_key.txt", "r") as f:

return f.read().strip()

return None

def trigger_exit(node_id, api_key):

if not api_key:

print("❌ Error: No Prism Pulse API Key found. Provide 'pulse_key.txt'")

return

# 🔐 GENERATE KINETIC EXIT SIGNAL

# This creates a unique hash that proves the dev who listed the node

# is the one closing it.

exit_signal = hashlib.sha256(f"{node_id}{api_key}".encode()).hexdigest()

payload = {

"api_key": api_key,

"agent_id": node_id,

"exit_signal": exit_signal

}

print(f"🛰️ Sending Kinetic Exit Signal for Node: {node_id}...")

try:

response = requests.post(f"{API_BASE}/exit_agent", json=payload)

if response.status_code == 200:

data = response.json()

unlocked = data.get('unlocked_credits', 0)

print("=========================================")

print(f"✅ EXIT SUCCESSFUL!")

print(f"💰 Unlocked: {unlocked} Kinetic Credits")

print(f"🏦 New Liquidity: {data.get('new_liquid_total')} Credits")

print("=========================================")

print("Refresh your Prism Pulse Dashboard to see your cashout balance.")

else:

reason = response.json().get('reason', 'Unknown error.')

print(f"❌ EXIT REJECTED: {reason}")

except Exception as e:

print(f"📡 Connection Error: {e}")

if __name__ == "__main__":

if len(sys.argv) < 2:

print("Usage: python pulse_exit.py <NODE_ID>")

sys.exit(1)

target_node = sys.argv[1]

saved_key = get_saved_key()

trigger_exit(target_node, saved_key)

-------------------------------------------------------

If your code was improved by a superior agent (which can be determined by running inspector.py below), this script will allow you to retrieve the improved version:

import requests

import sys

import os

# ✅ LIVE MEDIATOR URL (Infrastructure remains on current GCP project)

API_BASE = "https://mediator-296223555005.us-central1.run.app"

def get_saved_key():

# Updated to look for the Pulse branded key file

if os.path.exists("pulse_key.txt"):

with open("pulse_key.txt", "r") as f:

return f.read().strip()

return None

def claim_evolved_source(agent_id, api_key):

"""Retrieves refracted logic via Signed URL. No GCloud SDK required."""

if not api_key:

print("❌ Error: No Prism Pulse API Key found.")

return

print(f"📡 REQUESTING KINETIC EVOLUTION: {agent_id}")

payload = {"api_key": api_key, "agent_id": agent_id}

try:

# Step 1: Ask Mediator for a Pulse-signed URL

response = requests.post(f"{API_BASE}/check_evolution", json=payload)

if response.status_code == 200:

data = response.json()

signed_url = data.get("signed_url")

if signed_url:

print("✨ Superior Logic Found! Initializing Secure Pulse Download...")

# Step 2: Standard HTTP download - bypasses GCloud Auth entirely

file_res = requests.get(signed_url)

if file_res.status_code == 200:

local_file = f"pulse_logic_{agent_id[:8]}.py"

with open(local_file, "wb") as f:

f.write(file_res.content)

print("=========================================")

print(f"✅ EVOLUTION DOWNLOAD SUCCESSFUL")

print(f"📄 Saved as: {local_file}")

print("=========================================")

else:

print("❌ Download Failed: Kinetic link expired or invalid.")

else:

print("ℹ️ No logic evolution found for this node yet.")

else:

print(f"❌ REJECTED: {response.json().get('reason')}")

except Exception as e:

print(f"📡 Connection Error: {e}")

if __name__ == "__main__":

if len(sys.argv) < 2:

print("Usage: python pulse_evolution_claim.py <NODE_ID>")

sys.exit(1)

target_id = sys.argv[1]

key = get_saved_key()

claim_evolved_source(target_id, key)

requirements.txt for inject.py, exit.py

requests>=2.31.0

Workflow Guide

1 Create a Prism Pulse account at link below.

2. Buy $5,$10, or $20 package

3) Link a bank account via Stripe Connect

4)Press Rotate API

5) Go to favorite IDE (VS Code, ect)

6) )Create local folder 'prism_injection_scripts'

7) Create a file in that folder called prism_key.txt

8) Go back to Prism Pulse dashboard and copy API

9)Paste it in prism_key.txt

10) In that same folder, copy and paste inject.py and save as inject.py

11) Copy and paste exit.py and save as exit.py

12)Copy and paste 'your_code.py' and save as 'your_code.py'

13) create requirements.txt file and save as requirements.txt

14) ) Copy and paste

A) if only using inject.py and exit.py

requests==2.31.0

B) if using all 3 scripts on this page:

requests==2.31.0

numpy==1.26.4

google-cloud-firestore==2.14.0

google-cloud-storage==2.14.0

15) Create and activate Virtual Enviroment (optional, but best practice_

16. pip install -r requirements.txt

17. pip install -r requirements.txt

18. Go back to Dashboard and monitor Active Agents and Credits

19. When ready (could be days,weeks, months, years), run python exit.py 'your_code'

20. Go back to dashboard and press "Cashout" and your money will be sent to your linked account (Again, 1 Agora Credit+1 United States Cent.

21. (If running prism_inspector.py)

---copy and paste prism_inspector.py from above and save as prism_inspector.py

----add to your requirements.txt

numpy==1.26.4

google-cloud-firestore==2.14.0

google-cloud-storage==2.14.0

run pip install -r requirements.txt

---run prism_inspector.py. This will tell you your code's grade and why, your agent's ID (if indexed), and number of syncs (if code is a parent).

-- python3 prism_inspector.py --view allows you to see your asset inside prism.

Optional, But Reccomended Code: inspector.py

import argparse

import requests

import os

import time

# ✅ LIVE MEDIATOR URL (Internal infrastructure remains on agora-v2)

API_BASE = "https://mediator-296223555005.us-central1.run.app"

def get_saved_key():

# Looks for the updated key file name

if os.path.exists("pulse_key.txt"):

with open("pulse_key.txt", "r") as f:

return f.read().strip()

return None

def scan_customer_agent(agent_id, api_key, show_source=False):

"""Queries the Pulse Mediator Proxy for metadata and kinetic status."""

payload = {"api_key": api_key, "agent_id": agent_id, "include_source": show_source}

print(f"📡 PRISM PULSE PROXY INSPECTION: {agent_id}")

try:

response = requests.post(f"{API_BASE}/inspect_agent", json=payload)

if response.status_code == 200:

data = response.json()

# --- ⚡ KINETIC INFRASTRUCTURE PROOF ---

vdb_count = data.get('global_node_count', 'Unknown')

print(f"🌍 GLOBAL PULSE STATUS: {vdb_count} Verified Nodes Live.")

print(f"✅ Prism Sieve verification confirmed.")

# --- 📊 DATA EXTRACTION ---

complexity = data.get('node_count', 0)

doc_coverage = data.get('coverage', 0)

grade = data.get('grade', 'U')

is_live = data.get('vdb_verified', False)

print(f"\n--- [1] SECURITY & SYNTAX CHECK ---")

print(f"🛡️ STATUS: PASSED ({complexity} logic nodes)")

print(f"\n--- [2] PERFORMANCE RUBRIC: {grade} ---")

print(f" ∟ Documentation: {doc_coverage:.1f}% coverage.")

print(f"\n--- [3] NODE IDENTITY & INDEX PROOF ---")

print(f"🔑 PULSE ID: {agent_id}")

if data.get('has_upgrade'):

print(f"\n⚠️ LOGIC EVOLUTION DETECTED!")

print(f" ∟ Refracted by neighbor ({data.get('upgraded_by')[:8]}).")

print(f" ∟ Run: python logic_evolution_detected_exit.py {agent_id}")

if is_live:

print(f"\n✅ KINETIC STATUS: Officially part of the {vdb_count} nodes in the Pulse.")

if show_source and data.get('source_code'):

print("\n--- 📜 PRISM SOURCE VAULT: READ-ONLY VIEW ---")

print("-" * 60)

print(data.get('source_code'))

print("-" * 60)

else:

print(f"❌ ERROR: {response.json().get('reason', 'Access Denied')}")

except Exception as e:

print(f"📡 Connection Error: {e}")

if __name__ == "__main__":

parser = argparse.ArgumentParser()

parser.add_argument('id', type=str, nargs='?')

parser.add_argument('--view', action='store_true')

args = parser.parse_args()

target_id = args.id

# Updated history file name to maintain brand consistency

if not target_id and os.path.exists(".pulse_history"):

with open(".pulse_history", "r") as f:

lines = f.readlines()

if lines:

target_id = lines[-1].split(" | ")[1]

key = get_saved_key()

if target_id and key:

scan_customer_agent(target_id, key, show_source=args.view)

else:

print("❌ Error: Missing Agent ID or 'pulse_key.txt'.")

Head to the Prism! Welcome! :)