Protocol for Universal Language-based System Exchange
Universal semantic protocol for AI-to-AI communication. Think "TCP/IP for Artificial Intelligence."
🌍 Open Source & Free Forever | Apache 2.0 License Built for the community, by the community. Contributions welcome!
PULSE enables any AI system to communicate with any other AI system - regardless of vendor, framework, or architecture.
The Problem: Enterprises deploy 15-30 different AI systems that cannot communicate. Each integration costs $100K-$2M and takes 6-18 months.
The Solution: A universal semantic protocol with 1,000+ predefined concepts that eliminate ambiguity.
Instead of natural language (ambiguous, slow), PULSE uses semantic concepts:
- ❌ Natural: "Can you analyze the sentiment of this text?"
- ✅ PULSE:
ACT.ANALYZE.SENTIMENT+ENT.DATA.TEXT
Result: 1000× faster, 100% unambiguous, vendor-neutral communication.
- 🎯 Semantic Vocabulary - 120+ concepts (expanding to 1,000) across 10 categories
- 📝 JSON Encoding - Human-readable format for debugging and development
- ⚡ Binary Encoding - MessagePack format with 10× size reduction (Week 2 ✅)
- ✅ Automatic Validation - Validates against vocabulary with helpful error messages
- 🔒 Security Features - HMAC-SHA256 signing and replay protection (Week 3 ✅)
- 🖥️ CLI Tool - Command-line interface for all operations (Week 4 ✅)
- 📊 Type Safe - Full type hints for excellent IDE support
- 🧪 Well Tested - 165+ unit tests, 90%+ coverage
- 📖 Fully Documented - Comprehensive docstrings, examples, and guides
pip install pulse-protocol
# For development (with testing tools):
pip install pulse-protocol[dev]Requirements: Python 3.8+ | msgpack
from pulse import PulseMessage
# Create a message
message = PulseMessage(
action="ACT.QUERY.DATA",
target="ENT.DATA.TEXT",
parameters={"query": "hello world"}
)
# Serialize to JSON
json_output = message.to_json()
print(json_output)
# Serialize to binary (10× smaller)
binary_output = message.to_binary()
print(f"JSON size: {len(json_output)} bytes")
print(f"Binary size: {len(binary_output)} bytes")
# Deserialize from JSON
recreated = PulseMessage.from_json(json_output)
assert recreated.content["action"] == message.content["action"]
# Deserialize from binary
recreated_binary = PulseMessage.from_binary(binary_output)
assert recreated_binary.content["action"] == message.content["action"]from pulse import Vocabulary
# Validate a concept
is_valid = Vocabulary.validate_concept("ACT.ANALYZE.SENTIMENT")
print(f"Valid: {is_valid}") # True
# Search for concepts
results = Vocabulary.search("sentiment")
print(f"Found: {results}") # ['ACT.ANALYZE.SENTIMENT']
# Get concept details
description = Vocabulary.get_description("ACT.ANALYZE.SENTIMENT")
examples = Vocabulary.get_examples("ACT.ANALYZE.SENTIMENT")
print(f"{description}: {examples}")from pulse import PulseMessage, ValidationError
# Automatic validation
try:
message = PulseMessage(action="ACT.QUERY.DATA")
print("✓ Message valid!")
except ValidationError as e:
print(f"✗ Error: {e}")
# Invalid concept gets helpful suggestions
try:
message = PulseMessage(action="ACT.QUERY.INVALID")
except ValidationError as e:
print(f"Error: {e}")
# Output: "Invalid action concept: 'ACT.QUERY.INVALID'.
# Did you mean one of: ['ACT.QUERY.DATA', ...]?"
# Skip validation if needed
message = PulseMessage(action="CUSTOM.ACTION", validate=False)from pulse import PulseMessage, Encoder
# Create a message
message = PulseMessage(
action="ACT.ANALYZE.SENTIMENT",
target="ENT.DATA.TEXT",
parameters={"text": "PULSE is amazing!", "detail": "high"}
)
# Binary encoding (10× smaller than JSON)
binary_data = message.to_binary()
print(f"Binary size: {len(binary_data)} bytes") # ~80 bytes
# JSON for comparison
json_data = message.to_json(indent=None)
print(f"JSON size: {len(json_data)} bytes") # ~800 bytes
# Decode from binary
decoded = PulseMessage.from_binary(binary_data)
assert decoded.content["action"] == message.content["action"]
# Use unified Encoder for both formats
encoder = Encoder()
json_bytes = encoder.encode(message, format="json")
binary_bytes = encoder.encode(message, format="binary")
# Auto-detect format when decoding
decoded_msg = encoder.decode(binary_bytes) # Detects binary formatfrom pulse import PulseMessage, SecurityManager, KeyManager
# Initialize security manager with secret key
security = SecurityManager(secret_key="my-secret-key")
# Create and sign a message
message = PulseMessage(
action="ACT.TRANSFER.MONEY",
target="ENT.RESOURCE.DATABASE",
parameters={"amount": 1000, "to": "account-123"}
)
# Sign message with HMAC-SHA256
signature = security.sign_message(message)
print(f"Signature: {signature[:32]}...")
# Verify signature
is_valid = security.verify_signature(message)
print(f"Valid: {is_valid}") # True
# Tamper detection
message.content['parameters']['amount'] = 1000000
is_valid = security.verify_signature(message)
print(f"Valid after tampering: {is_valid}") # False - tampering detected!
# Replay protection
nonce_store = set()
result = security.check_replay_protection(message, nonce_store=nonce_store)
print(f"Replay check: {result['is_valid']}")
# Key management
km = KeyManager()
key = km.generate_and_store("agent-1")
retrieved_key = km.get_key("agent-1")# Create a message
$ pulse create --action ACT.QUERY.DATA --target ENT.DATA.TEXT -o message.json
# Validate message
$ pulse validate message.json
✓ Message is valid
# Sign message
$ pulse sign message.json --key my-secret-key -o signed.json
✓ Message signed
# Verify signature
$ pulse verify signed.json --key my-secret-key
✓ Signature is valid
# Encode to binary (10× smaller)
$ pulse encode message.json --format binary --compare
✓ Encoded to binary: message.bin
Size: 89 bytes
Size comparison:
JSON: 856 bytes
Binary: 89 bytes (9.6× smaller)
Savings: 89.6%
# Decode from binary
$ pulse decode message.bin -o decoded.json
✓ Decoded to: decoded.json
# See all commands
$ pulse --helpPULSE includes semantic concepts across 10 categories:
| Category | Count | Description | Example Concepts |
|---|---|---|---|
| ENT | 20+ | Entities - Data, agents, resources | ENT.DATA.TEXT, ENT.AGENT.AI, ENT.RESOURCE.DATABASE |
| ACT | 34+ | Actions - Operations | ACT.QUERY.DATA, ACT.ANALYZE.SENTIMENT, ACT.CREATE.TEXT |
| PROP | 16+ | Properties - Attributes | PROP.STATE.ACTIVE, PROP.QUALITY.HIGH, PROP.PRIORITY.HIGH |
| REL | 5+ | Relations - Relationships | REL.CONTAINS, REL.DEPENDS.ON, REL.RELATED.TO |
| LOG | 6+ | Logic - Operators | LOG.AND, LOG.OR, LOG.NOT, LOG.IF |
| MATH | 9+ | Mathematics - Operations | MATH.ADD, MATH.AVERAGE, MATH.SUM, MATH.COUNT |
| TIME | 6+ | Temporal - Time concepts | TIME.BEFORE, TIME.AFTER, TIME.NOW, TIME.FUTURE |
| SPACE | 6+ | Spatial - Space concepts | SPACE.INSIDE, SPACE.NEAR, SPACE.ABOVE, SPACE.BELOW |
| DATA | 7+ | Data Types - Structures | DATA.LIST, DATA.DICT, DATA.STRING, DATA.INTEGER |
| META | 11+ | Meta - Protocol control | META.STATUS.SUCCESS, META.ERROR.VALIDATION, META.REQUEST |
Total: 120+ concepts (expanding to 1,000 in future releases)
from pulse import PulseMessage
message = PulseMessage(
action="ACT.QUERY.DATA",
target="ENT.DATA.TEXT",
parameters={"query": "hello world"}
)
print(message.to_json())from pulse import PulseMessage
message = PulseMessage(
action="ACT.ANALYZE.SENTIMENT",
target="ENT.DATA.TEXT",
parameters={
"text": "I love PULSE Protocol!",
"detail_level": "PROP.DETAIL.HIGH"
}
)
# In a real system, this would be sent to a sentiment analysis agentfrom pulse import PulseMessage
message = PulseMessage(
action="ACT.QUERY.DATA",
target="ENT.RESOURCE.DATABASE",
parameters={
"table": "users",
"filters": {"status": "active"},
"limit": 10
}
)from pulse import PulseMessage, Encoder
message = PulseMessage(
action="ACT.PROCESS.BATCH",
target="ENT.DATA.TEXT",
parameters={"items": ["item1", "item2", "item3"]}
)
# Compare sizes
encoder = Encoder()
sizes = encoder.get_size_comparison(message)
print(f"JSON: {sizes['json']} bytes")
print(f"Binary: {sizes['binary']} bytes")
print(f"Reduction: {sizes['binary_reduction']}× smaller")
print(f"Savings: {sizes['savings_percent']}%")
# Typical output:
# JSON: 856 bytes
# Binary: 89 bytes
# Reduction: 9.6× smaller
# Savings: 89.6%from pulse import PulseMessage, ValidationError, EncodingError, DecodingError
# Handle validation errors
try:
message = PulseMessage(action="INVALID.ACTION")
except ValidationError as e:
print(f"Validation error: {e}")
# Use vocabulary search to find alternatives
results = Vocabulary.search("QUERY")
print(f"Did you mean: {results}")
# Retry with exponential backoff
max_retries = 5
base_delay = 0.1
for attempt in range(1, max_retries + 1):
try:
result = risky_operation()
break
except ConnectionError as e:
if attempt < max_retries:
delay = base_delay * (2 ** (attempt - 1))
time.sleep(delay)
else:
print("Max retries reached")from pulse import PulseMessage, SecurityManager, KeyManager
# Create security manager
security = SecurityManager(secret_key="my-secret-key")
# Sign a message
message = PulseMessage(
action="ACT.ANALYZE.SENTIMENT",
target="ENT.DATA.TEXT",
parameters={"text": "PULSE is secure!"}
)
signature = security.sign_message(message)
print(f"Signed: {signature[:32]}...")
# Verify signature
is_valid = security.verify_signature(message)
print(f"Valid: {is_valid}") # True
# Detect tampering
message.content['parameters']['text'] = "MODIFIED"
is_valid = security.verify_signature(message)
print(f"Valid after tampering: {is_valid}") # False!
# Replay protection
nonce_store = set()
result = security.check_replay_protection(message, nonce_store=nonce_store)
print(f"Age: {result['age_seconds']:.2f}s")
print(f"Valid: {result['is_valid']}")
# Typical output:
# Signed: a3f7b2c8...
# Valid: True
# Valid after tampering: False
# Age: 0.02s
# Valid: True# Create a message
$ pulse create --action ACT.QUERY.DATA --target ENT.DATA.TEXT \
--parameters '{"query": "test", "limit": 10}' \
-o message.json
# Validate the message
$ pulse validate message.json
✓ Message is valid
Action: ACT.QUERY.DATA
Type: REQUEST
# Sign with HMAC-SHA256
$ pulse sign message.json --key my-secret-key -o signed.json
✓ Message signed: signed.json
# Verify signature
$ pulse verify signed.json --key my-secret-key
✓ Signature is valid
Action: ACT.QUERY.DATA
# Encode to binary (10× smaller)
$ pulse encode signed.json --format binary --compare
✓ Encoded to binary: signed.bin
Size: 94 bytes
Size comparison:
JSON: 912 bytes
Binary: 94 bytes (9.7× smaller)
Savings: 89.7%
# Complete workflow automation
$ pulse create --action ACT.TRANSFER.MONEY --target ENT.RESOURCE.DATABASE \
--parameters '{"amount": 1000}' -o transfer.json && \
pulse sign transfer.json --key "$SECRET_KEY" -o transfer-signed.json && \
pulse verify transfer-signed.json --key "$SECRET_KEY" && \
pulse encode transfer-signed.json --format binary -o transfer.bin
# See all commands
$ pulse --helpSee examples/ for complete runnable examples:
01_hello_world.py- Basic message creation02_vocabulary_validation.py- Working with vocabulary03_use_cases.py- Real-world scenarios04_binary_encoding.py- Performance benchmarks ⚡05_error_handling.py- Error patterns and recovery ⚡06_security_features.py- Message signing and verification 🔒07_cli_usage.py- CLI tool demonstrations 🖥️
# Run all tests
pytest
# With coverage report
pytest --cov=pulse --cov-report=html --cov-report=term-missing
# Run specific test file
pytest tests/test_message.py
# Run with verbose output
pytest -v
# Run only unit tests (fast)
pytest -m unitTest Coverage: 165+ tests, 90%+ code coverage
Test Structure:
test_message.py- Core message functionalitytest_vocabulary.py- Vocabulary and concept validationtest_validator.py- Three-stage validation pipelinetest_encoder.py- Binary encoding, roundtrip, performance ⚡test_security.py- HMAC signing, replay protection 🔒test_cli.py- CLI commands and integration 🖥️
class PulseMessage:
"""Core PULSE Protocol message."""
def __init__(
self,
action: str, # Required: PULSE action concept
target: Optional[str] = None, # Optional: Target object concept
parameters: Optional[Dict] = None, # Optional: Parameters dict
sender: str = "default-agent", # Optional: Sender agent ID
validate: bool = True # Optional: Auto-validate on creation
)Methods:
to_json(indent=2) -> str- Serialize to JSON stringto_binary() -> bytes- Serialize to binary MessagePack format (10× smaller) ⚡to_dict() -> dict- Convert to dictionaryfrom_json(json_str) -> PulseMessage- Class method to deserialize from JSONfrom_binary(binary_data) -> PulseMessage- Class method to deserialize from binary ⚡validate(check_freshness=False) -> bool- Validate message
Attributes:
envelope: dict- Message metadata (version, timestamp, sender, receiver, message_id, nonce, signature)type: str- Message type (REQUEST, RESPONSE, ERROR, STATUS)content: dict- Message payload (action, object, parameters)
class Vocabulary:
"""PULSE vocabulary management."""
# Validation
@classmethod
def validate_concept(cls, concept: str) -> bool
# Search
@classmethod
def search(cls, query: str) -> List[str]
# Information
@classmethod
def get_category(cls, concept: str) -> Optional[str]
@classmethod
def get_description(cls, concept: str) -> Optional[str]
@classmethod
def get_examples(cls, concept: str) -> List[str]
# Organization
@classmethod
def list_by_category(cls, category: str) -> List[str]
@classmethod
def count_by_category(cls) -> Dict[str, int]
@classmethod
def get_all_categories(cls) -> Set[str]class MessageValidator:
"""Validate PULSE messages."""
@staticmethod
def validate_message(message, check_freshness=True) -> bool
@staticmethod
def validate_envelope(envelope: dict) -> bool
@staticmethod
def validate_content(content: dict) -> bool
@staticmethod
def validate_timestamp_freshness(
timestamp: str,
max_age_seconds: int = 300
) -> boolfrom pulse import Encoder, JSONEncoder, BinaryEncoder
# Unified Encoder (recommended)
encoder = Encoder()
json_bytes = encoder.encode(message, format="json")
binary_bytes = encoder.encode(message, format="binary")
decoded = encoder.decode(binary_bytes) # Auto-detects format
# Get size comparison
sizes = encoder.get_size_comparison(message)
print(f"JSON: {sizes['json']} bytes")
print(f"Binary: {sizes['binary']} bytes ({sizes['binary_reduction']}× smaller)")
print(f"Savings: {sizes['savings_percent']}%")
# JSONEncoder - Human-readable format
json_encoder = JSONEncoder()
json_bytes = json_encoder.encode(message)
decoded = json_encoder.decode(json_bytes)
# BinaryEncoder - MessagePack format (10× smaller)
binary_encoder = BinaryEncoder()
binary_bytes = binary_encoder.encode(message)
decoded = binary_encoder.decode(binary_bytes)Encoder Methods:
encode(message, format="json") -> bytes- Encode in specified formatdecode(data, format=None) -> PulseMessage- Decode (auto-detects if format not specified)get_size_comparison(message) -> dict- Compare sizes across formats
Available Formats:
"json"- Human-readable, ~800 bytes typical"binary"- MessagePack, ~80 bytes (10× reduction) ⚡"compact"- Custom format, ~60 bytes (13× reduction) - Coming soon
from pulse import SecurityManager, KeyManager
# SecurityManager - HMAC-SHA256 signing and verification
security = SecurityManager(secret_key="my-secret-key")
# Sign message
message = PulseMessage(action="ACT.QUERY.DATA")
signature = security.sign_message(message)
# Verify signature
is_valid = security.verify_signature(message)
# Check replay protection
result = security.check_replay_protection(
message,
max_age_seconds=300, # 5 minutes
nonce_store=set() # For nonce deduplication
)
# KeyManager - Simple key storage
km = KeyManager()
key = km.generate_and_store("agent-1")
retrieved = km.get_key("agent-1")SecurityManager Methods:
sign_message(message) -> str- Sign message with HMAC-SHA256verify_signature(message, expected_signature=None) -> bool- Verify signaturecheck_replay_protection(message, max_age_seconds=300, nonce_store=None) -> dict- Check replay indicatorsgenerate_key() -> str- Static method to generate secure random key
KeyManager Methods:
generate_and_store(agent_id) -> str- Generate and store key for agentstore_key(agent_id, key)- Store existing keyget_key(agent_id) -> Optional[str]- Retrieve stored keyremove_key(agent_id) -> bool- Remove stored keylist_agents() -> list- List all agents with keys
Security Features:
- HMAC-SHA256 message signing
- Constant-time signature comparison (timing attack protection)
- Replay protection (timestamp freshness + nonce deduplication)
- Tamper detection (any modification invalidates signature)
- Performance: ~1-2ms per operation
# Clone repository
git clone https://github.com/pulseprotocolorg-cyber/pulse-python.git
cd pulse-python
# Create virtual environment
python -m venv venv
source venv/bin/activate # Windows: venv\Scripts\activate
# Install dependencies
pip install -r requirements.txt
pip install -r requirements-dev.txt
pip install -e .# Format code
black pulse/ tests/
# Lint code
pylint pulse/
# Type check
mypy pulse/
# Run all checks
black pulse/ tests/ && pylint pulse/ && mypy pulse/ && pytestpulse-python/
├── pulse/ # Main package
│ ├── __init__.py # Package exports
│ ├── message.py # PulseMessage class
│ ├── vocabulary.py # Vocabulary system (120+ concepts)
│ ├── validator.py # MessageValidator
│ ├── encoder.py # JSON/Binary/Compact encoders ⚡
│ ├── security.py # SecurityManager, KeyManager 🔒
│ ├── cli.py # Command-line interface 🖥️
│ ├── benchmarks.py # Performance benchmarks 🖥️
│ ├── exceptions.py # Custom exceptions
│ └── version.py # Version info
├── tests/ # Test suite (165+ tests)
│ ├── test_message.py
│ ├── test_vocabulary.py
│ ├── test_validator.py
│ ├── test_encoder.py # Binary encoding tests ⚡
│ ├── test_security.py # Security tests 🔒
│ └── test_cli.py # CLI tests 🖥️
├── examples/ # Usage examples
│ ├── 01_hello_world.py
│ ├── 02_vocabulary_validation.py
│ ├── 03_use_cases.py
│ ├── 04_binary_encoding.py ⚡
│ ├── 05_error_handling.py ⚡
│ ├── 06_security_features.py 🔒
│ └── 07_cli_usage.py 🖥️
└── docs/ # Documentation
We welcome contributions! See CONTRIBUTING.md for guidelines.
Quick Start:
- Fork the repository
- Create a feature branch (
git checkout -b feature/amazing-feature) - Write tests for your changes
- Ensure all tests pass and code is formatted (
black,pylint,mypy) - Commit your changes (
git commit -m 'feat: add amazing feature') - Push to your branch (
git push origin feature/amazing-feature) - Open a Pull Request
Apache License 2.0 - See LICENSE for details.
This project is open source and will remain free forever.
Version: 0.5.0 License: Apache 2.0 - Free & Open Source Forever 🌍 Python: 3.8+ Status: Active Development - Ready for Contributors!
💡 This project is production-ready for early adopters. We welcome contributions from developers worldwide!
- Core message creation and parsing
- JSON encoding/decoding (human-readable)
- Binary encoding/decoding (MessagePack, 10× size reduction) ⚡
- HMAC-SHA256 message signing for integrity 🔒
- Replay protection (timestamp + nonce deduplication) 🔒
- Tamper detection and signature verification 🔒
- CLI tool (create, validate, sign, verify, encode, decode) 🖥️
- Performance benchmarks with statistical analysis 🖥️
- Vocabulary system (120+ concepts across 10 categories)
- Three-stage message validation
- Error handling patterns (retry, circuit breaker, graceful degradation)
- Unified Encoder with auto-format detection
- Key management (SecurityManager, KeyManager)
- 165+ unit tests with 90%+ coverage
- Future: Compact encoding (13× reduction), TLS integration, network client/server, framework integrations, 1,000 concepts
- Vocabulary contains 120 concepts (target: 1,000)
- Compact encoding not yet implemented (placeholder in place)
- TLS integration not yet implemented (Week 4)
- No network transport yet (Week 4)
Problem: AI systems can't talk to each other
- 15-30 different AI systems per enterprise
- Each integration costs $100K-$2M
- Takes 6-18 months per integration
- Result: Digital Tower of Babel
Solution: Universal protocol
- ✅ Vendor neutral (works with any AI)
- ✅ 1000× faster than natural language
- ✅ 100% unambiguous communication
- ✅ Open source, forever free
- ✅ Evolutionarily stable design
- Issues: GitHub Issues
- Discussions: GitHub Discussions
- Email: dev@pulse-protocol.org
PULSE Protocol is free and open source forever. If you find it useful, consider supporting the development:
| Currency | Address |
|---|---|
| BTC | bc1qawmyg0merz7027q0s74lgret6aaldswgk43r7z |
| ETH | 0xf39be73240a32397E9004a3c0dbC8f63E52C724B |
EUR:
- Name: Sergej Klein
- IBAN:
BE59 9675 3051 8426 - SWIFT/BIC:
TRWIBEB1XXX
USD:
- Name: Sergej Klein
- Account:
985160876270679 - Routing (Wire/ACH):
084009519 - SWIFT/BIC:
TRWIUS35XXX
All donations go directly to project development and infrastructure.
Built with ❤️ by the PULSE Protocol team
Let's build the future of AI communication together.