pulse-protocol 0.5.0

Universal semantic protocol for AI-to-AI communication

PULSE Protocol - Python Implementation

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!

---

🎯 What is PULSE?

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.

Key Innovation

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.

---

✨ Features

• 🎯 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

---

🚀 Quick Start

Installation

# Coming soon to PyPI!
# pip install pulse-protocol

# For now - Install from GitHub:
git clone https://github.com/pulse-protocol/pulse-python.git
cd pulse-python
pip install -e .

# Or install dependencies only:
pip install msgpack>=1.0.0

# For development (with testing tools):
pip install -e ".[dev]"

Requirements: Python 3.8+ | msgpack

Basic Usage

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"]

Using the Vocabulary

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}")

Message Validation

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)

Binary Encoding (Week 2 ✅)

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 format

Security Features (Week 3 ✅)

from 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")

CLI Tool (Week 4 ✅)

# 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 --help

---

📊 Vocabulary Categories

PULSE 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)

---

📖 Examples

Example 1: Hello World

from pulse import PulseMessage

message = PulseMessage(
    action="ACT.QUERY.DATA",
    target="ENT.DATA.TEXT",
    parameters={"query": "hello world"}
)

print(message.to_json())

Example 2: Sentiment Analysis

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 agent

Example 3: Database Query

from pulse import PulseMessage

message = PulseMessage(
    action="ACT.QUERY.DATA",
    target="ENT.RESOURCE.DATABASE",
    parameters={
        "table": "users",
        "filters": {"status": "active"},
        "limit": 10
    }
)

Example 4: Binary Encoding ⚡

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%

Example 5: Error Handling

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")

Example 6: Security Features 🔒

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

Example 7: CLI Tool 🖥️

# 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 --help

See examples/ for complete runnable examples:

• 01_hello_world.py - Basic message creation
• 02_vocabulary_validation.py - Working with vocabulary
• 03_use_cases.py - Real-world scenarios
• 04_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 🖥️

---

🧪 Running Tests

# 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 unit

Test Coverage: 165+ tests, 90%+ code coverage

Test Structure:

• test_message.py - Core message functionality
• test_vocabulary.py - Vocabulary and concept validation
• test_validator.py - Three-stage validation pipeline
• test_encoder.py - Binary encoding, roundtrip, performance ⚡
• test_security.py - HMAC signing, replay protection 🔒
• test_cli.py - CLI commands and integration 🖥️

---

📚 API Reference

PulseMessage

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 string
• to_binary() -> bytes - Serialize to binary MessagePack format (10× smaller) ⚡
• to_dict() -> dict - Convert to dictionary
• from_json(json_str) -> PulseMessage - Class method to deserialize from JSON
• from_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)

Vocabulary

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]

MessageValidator

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
    ) -> bool

Encoder Classes ⚡

from 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 format
• decode(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

Security Classes 🔒

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-SHA256
• verify_signature(message, expected_signature=None) -> bool - Verify signature
• check_replay_protection(message, max_age_seconds=300, nonce_store=None) -> dict - Check replay indicators
• generate_key() -> str - Static method to generate secure random key

KeyManager Methods:

• generate_and_store(agent_id) -> str - Generate and store key for agent
• store_key(agent_id, key) - Store existing key
• get_key(agent_id) -> Optional[str] - Retrieve stored key
• remove_key(agent_id) -> bool - Remove stored key
• list_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

---

🛠️ Development

Setup Development Environment

# Clone repository
git clone https://github.com/pulse-protocol/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 .

Code Quality Tools

# Format code
black pulse/ tests/

# Lint code
pylint pulse/

# Type check
mypy pulse/

# Run all checks
black pulse/ tests/ && pylint pulse/ && mypy pulse/ && pytest

Project Structure

pulse-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

---

🤝 Contributing

We welcome contributions! See CONTRIBUTING.md for guidelines.

Quick Start:

1. Fork the repository
2. Create a feature branch (git checkout -b feature/amazing-feature)
3. Write tests for your changes
4. Ensure all tests pass and code is formatted (black, pylint, mypy)
5. Commit your changes (git commit -m 'feat: add amazing feature')
6. Push to your branch (git push origin feature/amazing-feature)
7. Open a Pull Request

---

📄 License

Apache License 2.0 - See LICENSE for details.

This project is open source and will remain free forever.

---

📊 Project Status

Version: 0.4.0 (Alpha - Week 4 Complete ✅) 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!

What's Working ✅

• 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

Coming Soon 🚧

• Future: Compact encoding (13× reduction), TLS integration, network client/server, framework integrations, 1,000 concepts

Known Limitations

• 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)

---

🌟 Why PULSE?

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

---

📞 Support & Community

• Issues: GitHub Issues
• Discussions: GitHub Discussions
• Email: [e-mail]

---

Support the Project

PULSE Protocol is free and open source forever. If you find it useful, consider supporting the development:

Crypto

Currency	Address
BTC	bc1qawmyg0merz7027q0s74lgret6aaldswgk43r7z
ETH	0xf39be73240a32397E9004a3c0dbC8f63E52C724B

Bank Transfer (Wise)

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.