Skip to main content

Semantic Kernel integration for xLink identity-based authentication (orchestration + planning + memory, 603× faster M2M auth)

Project description

Private.Me xLink for Semantic Kernel (Python)

Identity-based authentication for Semantic Kernel plugins and orchestration.

Python bindings for @private.me/semantic-kernel-xlink. Build Semantic Kernel AI applications with cryptographic identity instead of API keys. Zero-config setup, enterprise-grade access control, no cascading failures.

Installation

Prerequisites

This package requires both the Node.js module and Python bindings:

# 1. Install Node.js package
npm install @private.me/semantic-kernel-xlink semantic-kernel

# 2. Install Python bindings
pip install private-me-semantic-kernel-xlink

Requirements:

  • Python 3.9+
  • Node.js 18+ (for backend)
  • npm (for Node.js package installation)
  • Semantic Kernel SDK

Quick Start

from private_me import semantic_kernel_xlink

# Create kernel plugin (generates identity automatically)
plugin = await semantic_kernel_xlink.create_plugin(
    name='my-plugin',
    verbose=True
)

# Execute skill with identity auth (no credentials!)
result = await plugin.call(
    skill='text-summarization',
    params={'text': 'Long document to summarize...'}
)

if result.ok:
    print(f"Summary: {result.value}")

That's it. No API keys, no OAuth, no configuration files.

API Reference

create_plugin(name: str, policy: Dict[str, Any] = None, verbose: bool = False) -> SemanticKernelXLinkPlugin

Create Semantic Kernel plugin with identity-based authentication.

Parameters:

  • name (str): Plugin name for identification
  • policy (Dict[str, Any], optional): Policy configuration
    • allowed_skills (List[str]): Whitelist of allowed skills
    • max_plugin_calls (int): Maximum plugin calls before refresh
    • execution_timeout (int): Timeout in milliseconds (default: 30000)
  • verbose (bool): Enable logging (default: False)

Returns:

  • SemanticKernelXLinkPlugin: Plugin instance with cryptographic identity

Example:

plugin = await semantic_kernel_xlink.create_plugin(
    name='research-agent',
    policy={
        'allowed_skills': ['summarization', 'text-analysis'],
        'max_plugin_calls': 100,
        'execution_timeout': 30000
    },
    verbose=True
)

SemanticKernelXLinkPlugin.get_did() -> str

Get plugin's decentralized identifier.

Returns:

  • str: DID (e.g., did:key:z6Mk...)

Example:

did = plugin.get_did()
print(f"Plugin Identity: {did}")

SemanticKernelXLinkPlugin.call(skill: str, params: Dict[str, Any]) -> Result

Execute Semantic Kernel skill with identity authentication.

Parameters:

  • skill (str): Skill name to execute
  • params (Dict[str, Any]): Skill parameters

Returns:

  • Result: Result object with .ok (bool) and .value or .error

Raises:

  • RuntimeError: If Node.js backend fails
  • PolicyViolationError: If skill execution violates policy

Example:

result = await plugin.call(
    skill='text-summarization',
    params={'text': 'Research paper abstract...', 'max_length': 100}
)

if result.ok:
    print(f"Summary: {result.value}")
else:
    print(f"Error: {result.error['title']}")

SemanticKernelXLinkPlugin.send(to: str, payload: Dict[str, Any]) -> Result

Send message to another agent using xLink identity.

Parameters:

  • to (str): Recipient's DID
  • payload (Dict[str, Any]): Message payload

Returns:

  • Result: Result object with .ok (bool)

Raises:

  • RuntimeError: If message delivery fails

Example:

await plugin.send(
    to='did:key:z6Mkrecipient...',
    payload={
        'type': 'task-complete',
        'results': ['Finding 1', 'Finding 2']
    }
)

SemanticKernelXLinkPlugin.receive() -> Result

Receive messages sent to this plugin.

Returns:

  • Result: Result object containing list of messages

Example:

messages = await plugin.receive()
if messages.ok:
    for msg in messages.value:
        print(f"From: {msg['from']}")
        print(f"Payload: {msg['payload']}")

SemanticKernelXLinkPlugin.get_audit_log() -> List[Dict[str, Any]]

Retrieve execution audit trail for compliance and monitoring.

Returns:

  • List[Dict[str, Any]]: Audit log entries with timestamps, skills, status

Example:

audit_log = plugin.get_audit_log()
for entry in audit_log:
    print(f"[{entry['timestamp']}] {entry['skill']}: {entry['status']}")

SemanticKernelXLinkPlugin.export_identity() -> Result

Export identity for persistence across sessions.

Returns:

  • Result: Result containing PKCS8 private key bytes

Example:

export_result = await plugin.export_identity()
if export_result.ok:
    with open('plugin-identity.key', 'wb') as f:
        f.write(export_result.value)

from_identity(pkcs8: bytes, name: str, policy: Dict[str, Any] = None) -> Result

Create plugin from existing identity.

Parameters:

  • pkcs8 (bytes): PKCS8 private key
  • name (str): Plugin name
  • policy (Dict[str, Any], optional): Policy configuration

Returns:

  • Result: Result containing plugin with restored identity

Example:

with open('plugin-identity.key', 'rb') as f:
    identity_bytes = f.read()

restore_result = await semantic_kernel_xlink.from_identity(
    pkcs8=identity_bytes,
    name='restored-plugin'
)

if restore_result.ok:
    restored_plugin = restore_result.value
    print(restored_plugin.get_did())  # Same DID as original

Usage Examples

Basic Skill Execution

from private_me import semantic_kernel_xlink

# Create plugin
plugin = await semantic_kernel_xlink.create_plugin(name='assistant')

print(f"Plugin Identity: {plugin.get_did()}")

# Execute summarization skill
result = await plugin.call(
    skill='text-summarization',
    params={'text': 'Semantic Kernel is a lightweight SDK...'}
)

if result.ok:
    print(f"Summary: {result.value}")
else:
    print(f"Error: {result.error['detail']}")

Policy-Enforced Plugin

from private_me import semantic_kernel_xlink

# Create plugin with strict policy
plugin = await semantic_kernel_xlink.create_plugin(
    name='policy-enforced',
    policy={
        'allowed_skills': ['summarization', 'text-analysis'],
        'max_plugin_calls': 50,
        'execution_timeout': 30000
    }
)

# Allowed skill - succeeds
result = await plugin.call(
    skill='summarization',
    params={'text': 'Document content...'}
)
print(f"Allowed: {result.ok}")  # True

# Disallowed skill - policy violation
result = await plugin.call(
    skill='image-generation',
    params={'prompt': 'sunset'}
)
print(f"Blocked: {not result.ok}")  # True
print(f"Reason: {result.error['detail']}")  # "Skill not in allowedSkills list"

Multi-Agent Orchestration

from private_me import semantic_kernel_xlink

# Create orchestrator and worker agents
orchestrator = await semantic_kernel_xlink.create_plugin(name='orchestrator')
worker = await semantic_kernel_xlink.create_plugin(name='worker')

# Orchestrator assigns task to worker
await orchestrator.send(
    to=worker.get_did(),
    payload={
        'type': 'task-assignment',
        'task': 'analyze-data',
        'data': {'records': 1000}
    }
)

# Worker receives and processes task
messages = await worker.receive()
if messages.ok and len(messages.value) > 0:
    task = messages.value[0]['payload']
    print(f"Received task: {task['task']}")

    # Execute skill
    result = await worker.call(
        skill='data-analysis',
        params={'records': task['data']['records']}
    )

    # Send results back
    await worker.send(
        to=orchestrator.get_did(),
        payload={
            'type': 'task-complete',
            'results': result.value
        }
    )

Identity Persistence

from private_me import semantic_kernel_xlink
import os

# Create plugin and export identity
plugin = await semantic_kernel_xlink.create_plugin(name='persistent-agent')
original_did = plugin.get_did()

export_result = await plugin.export_identity()
if export_result.ok:
    with open('agent-identity.key', 'wb') as f:
        f.write(export_result.value)

# Later: restore plugin with same identity
with open('agent-identity.key', 'rb') as f:
    identity_bytes = f.read()

restore_result = await semantic_kernel_xlink.from_identity(
    pkcs8=identity_bytes,
    name='persistent-agent'
)

if restore_result.ok:
    restored_plugin = restore_result.value
    print(original_did == restored_plugin.get_did())  # True

Audit Trail for Compliance

from private_me import semantic_kernel_xlink

# Create plugin with audit enabled
plugin = await semantic_kernel_xlink.create_plugin(
    name='audited-agent',
    verbose=True
)

# Execute multiple skills
await plugin.call('summarization', {'text': 'Doc 1'})
await plugin.call('text-analysis', {'text': 'Doc 2'})
await plugin.call('translation', {'text': 'Doc 3', 'target': 'es'})

# Retrieve audit log
audit_log = plugin.get_audit_log()
print(f"Total executions: {len(audit_log)}")

for entry in audit_log:
    print(f"[{entry['timestamp']}] Skill: {entry['skill']}")
    print(f"  Status: {entry['status']}")
    print(f"  Duration: {entry['duration_ms']}ms")

Error Handling with RFC 7807

from private_me import semantic_kernel_xlink

plugin = await semantic_kernel_xlink.create_plugin(name='agent')

result = await plugin.call(
    skill='text-summarization',
    params={'text': 'Too short'}  # Invalid input
)

if not result.ok:
    # Structured error with field-level details
    error = result.error

    print(f"Error Type: {error['type']}")
    print(f"Title: {error['title']}")
    print(f"Detail: {error['detail']}")
    print(f"Instance: {error['instance']}")

    if 'invalid_params' in error:
        print("Invalid parameters:")
        for param in error['invalid_params']:
            print(f"  - {param['name']}: {param['reason']}")

    if 'fix' in error:
        print(f"Fix: {error['fix']}")

Policy Configuration

Control plugin behavior with fine-grained policies:

Policy Field Type Description
allowed_skills List[str] Whitelist of executable skills
max_plugin_calls int Maximum calls before identity refresh
execution_timeout int Timeout per skill call (milliseconds)

Example:

policy = {
    'allowed_skills': ['summarization', 'translation'],
    'max_plugin_calls': 100,
    'execution_timeout': 60000  # 1 minute
}

plugin = await semantic_kernel_xlink.create_plugin(
    name='policy-agent',
    policy=policy
)

Architecture

This package uses a wrapper pattern:

Python App → Python Bindings → Node.js Backend → Semantic Kernel SDK
  1. Python layer: Provides Pythonic API (create_plugin(), call())
  2. Node.js backend: Handles cryptographic operations (Ed25519, DID generation)
  3. Semantic Kernel integration: Plugin execution, policy enforcement, audit logging

Troubleshooting

"Node.js module not found"

Error:

RuntimeError: Node.js module @private.me/semantic-kernel-xlink not found

Solution:

# Install Node.js package first
npm install @private.me/semantic-kernel-xlink semantic-kernel

# Verify installation
ls node_modules/@private.me/semantic-kernel-xlink

"Node.js backend error"

Error:

RuntimeError: Node.js backend error: <message>

Solution:

  • Check Node.js version (requires 18+): node --version
  • Verify Node.js package installed: npm list @private.me/semantic-kernel-xlink
  • Check Semantic Kernel installed: npm list semantic-kernel
  • Review Node.js error message for details

Policy violation errors

Error:

PolicyViolationError: Skill not in allowedSkills list

Solution:

  • Verify skill name matches exactly (case-sensitive)
  • Add skill to allowed_skills list in policy configuration
  • Check skill is available in Semantic Kernel installation
  • Review audit log to see which skills are being called

Skill execution timeout

Error:

RuntimeError: Skill execution timeout after 30000ms

Solution:

  • Increase execution_timeout in policy configuration
  • Optimize skill parameters (reduce input size, lower max tokens)
  • Check network connectivity to LLM provider
  • Review skill implementation for performance issues

Invalid identity errors

Error:

ValueError: Invalid PKCS8 private key format

Solution:

  • Verify exported identity is complete (not truncated)
  • Check file permissions allow reading identity file
  • Ensure identity was exported via export_identity() method
  • Verify identity bytes are not corrupted (checksum validation)

Message delivery failures

Error:

RuntimeError: Message delivery failed: recipient not found

Solution:

  • Verify recipient DID is valid (starts with did:key:)
  • Ensure recipient plugin is running and connected
  • Check network connectivity between agents
  • Review xLink registry configuration

Development

Running Tests

# Install dev dependencies
pip install -e ".[dev]"

# Run tests
pytest -v

# Run with coverage
pytest -v --cov=private_me --cov-report=html

Building

# Build wheel
python setup.py bdist_wheel

# Validate build
bash validate-build.sh

Support

License

Proprietary - See LICENSE.md


Questions? Visit private.me/docs/semantic-kernel-xlink for complete documentation.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

private_me_semantic_kernel_xlink-0.1.1.tar.gz (17.1 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page