Python Wrapper

The JuliaOS Python Wrapper provides a comprehensive Python interface to the JuliaOS backend. It allows you to create and manage agents, swarms, wallets, and more from Python applications.

Installation

pip install -e /path/to/JuliaOS/packages/python-wrapper

For additional features, you can install optional dependencies:

# With LLM support
pip install -e /path/to/JuliaOS/packages/python-wrapper[llm]

# With Google ADK support
pip install -e /path/to/JuliaOS/packages/python-wrapper[adk]

# With development tools
pip install -e /path/to/JuliaOS/packages/python-wrapper[dev]

Overview

The Python wrapper connects to the Julia backend through a WebSocket bridge and provides an async/await-based API for interacting with JuliaOS. It includes support for:

• Creating and managing agents

• Creating and managing swarms

• Wallet management

• Blockchain integration

• Storage operations

• LangChain integration

• NumPy integration for scientific computing and optimization

• Async context manager support for cleaner code

• Comprehensive swarm algorithm implementations (DE, PSO, GWO, ACO, GA, WOA, DEPSO)

Basic Usage

import asyncio
from juliaos import JuliaOS
from juliaos.agents import AgentType

async def main():
    # Initialize JuliaOS using async context manager
    async with JuliaOS(host="localhost", port=8052) as juliaos_client:
        # Create an agent
        agent = await juliaos_client.agents.create_agent(
            name="MyAgent",
            agent_type=AgentType.TRADING,
            config={
                "parameters": {
                    "risk_tolerance": 0.5,
                    "max_position_size": 1000.0
                }
            }
        )

        # Start the agent
        await agent.start()

        # Execute a task with the agent
        task_result = await agent.execute_task({
            "action": "analyze_market",
            "market": "crypto",
            "timeframe": "1d"
        })

        print("Task result:", task_result)
    # Connection is automatically closed when exiting the context

if __name__ == "__main__":
    asyncio.run(main())

Agents

The agents module provides classes and functions for creating and managing agents:

import asyncio
from juliaos import JuliaOS
from juliaos.agents import AgentType

async def agent_example():
    juliaos_client = JuliaOS(host="localhost", port=8052)
    await juliaos_client.connect()

    # Create an agent
    agent = await juliaos_client.agents.create_agent(
        name="MyAgent",
        agent_type=AgentType.TRADING,
        config={
            "parameters": {
                "risk_tolerance": 0.5,
                "max_position_size": 1000.0
            }
        }
    )

    # Start the agent
    await agent.start()

    # Get agent status
    status = await agent.get_status()
    print("Agent status:", status)

    # Execute a task
    task_result = await agent.execute_task({
        "action": "analyze_market",
        "market": "crypto",
        "timeframe": "1d"
    })
    print("Task result:", task_result)

    # Stop the agent
    await agent.stop()

    # List all agents
    agents = await juliaos_client.agents.list_agents()
    print("All agents:", agents)

    # Get a specific agent
    retrieved_agent = await juliaos_client.agents.get_agent(agent.id)
    print("Retrieved agent:", retrieved_agent)

    # Delete the agent
    await agent.delete()

Swarms

The swarms module provides classes and functions for creating and managing swarms:

import asyncio
from juliaos import JuliaOS
from juliaos.swarms import SwarmType

async def swarm_example():
    # Initialize JuliaOS using async context manager
    async with JuliaOS(host="localhost", port=8052) as juliaos_client:
        # Create a swarm
        swarm = await juliaos_client.swarms.create_swarm(
            name="MySwarm",
            swarm_type=SwarmType.OPTIMIZATION,
            algorithm="PSO",
            dimensions=10,
            bounds=[(-10.0, 10.0)] * 10,
            config={
                "particles": 30,
                "c1": 2.0,
                "c2": 2.0,
                "w": 0.7
            }
        )

        # Start the swarm
        await swarm.start()

        # Get swarm status
        status = await swarm.get_status()
        print("Swarm status:", status)

        # Register an objective function
        function_id = "sphere"
        await juliaos_client.swarms.set_objective_function(
            function_id=function_id,
            function_code="function(x) return sum(x.^2) end",
            function_type="julia"
        )

        # Run optimization
        optimization_result = await swarm.run_optimization(
            function_id=function_id,
            max_iterations=100,
            max_time_seconds=60,
            tolerance=1e-6
        )
        print("Optimization result:", optimization_result)

        # Stop the swarm
        await swarm.stop()

        # List all swarms
        swarms = await juliaos_client.swarms.list_swarms()
        print("All swarms:", swarms)

        # Get a specific swarm
        retrieved_swarm = await juliaos_client.swarms.get_swarm(swarm.id)
        print("Retrieved swarm:", retrieved_swarm)

        # Delete the swarm
        await swarm.delete()

Specialized Agent Types

The Python wrapper supports specialized agent types:

Trading Agents

import asyncio
from juliaos import JuliaOS
from juliaos.agents import AgentType
from juliaos.agents.specialized import TradingAgent

async def trading_agent_example():
    juliaos_client = JuliaOS(host="localhost", port=8052)
    await juliaos_client.connect()

    # Create a trading agent
    trading_agent = await TradingAgent.create(
        juliaos_client,
        name="MyTradingAgent",
        risk_level="medium",
        max_position_size=1000.0,
        take_profit=0.05,
        stop_loss=0.03,
        trading_pairs=["ETH/USDC", "BTC/USDC"],
        strategies=["momentum", "mean_reversion"]
    )

    # Start the agent
    await trading_agent.start()

    # Execute a trade
    trade_result = await trading_agent.execute_trade(
        pair="ETH/USDC",
        side="buy",
        amount=0.1,
        price=2000.0,
        type="limit"
    )
    print("Trade result:", trade_result)

    # Get portfolio
    portfolio = await trading_agent.get_portfolio()
    print("Portfolio:", portfolio)

    # Get trade history
    trade_history = await trading_agent.get_trade_history(limit=10)
    print("Trade history:", trade_history)

    # Stop the agent
    await trading_agent.stop()

Research Agents

import asyncio
from juliaos import JuliaOS
from juliaos.agents import AgentType
from juliaos.agents.specialized import ResearchAgent

async def research_agent_example():
    juliaos_client = JuliaOS(host="localhost", port=8052)
    await juliaos_client.connect()

    # Create a research agent
    research_agent = await ResearchAgent.create(
        juliaos_client,
        name="MyResearchAgent",
        research_areas=["market", "technology", "sentiment"],
        data_sources=["web", "api", "database"],
        analysis_methods=["statistical", "nlp", "trend"],
        output_formats=["text", "json", "chart"]
    )

    # Start the agent
    await research_agent.start()

    # Conduct research
    research_result = await research_agent.conduct_research(
        topic="Ethereum Layer 2 Solutions",
        depth="medium",
        focus=["technology", "adoption", "performance"],
        timeframe="last_6_months",
        sources=["academic", "news", "social_media"]
    )
    print("Research result:", research_result)

    # Get research history
    research_history = await research_agent.get_research_history(limit=10)
    print("Research history:", research_history)

    # Stop the agent
    await research_agent.stop()

Dev Agents

import asyncio
from juliaos import JuliaOS
from juliaos.agents import AgentType
from juliaos.agents.specialized import DevAgent

async def dev_agent_example():
    juliaos_client = JuliaOS(host="localhost", port=8052)
    await juliaos_client.connect()

    # Create a dev agent
    dev_agent = await DevAgent.create(
        juliaos_client,
        name="MyDevAgent",
        languages=["python", "javascript", "julia"],
        frameworks=["react", "tensorflow", "flask"],
        specialties=["web", "ai", "blockchain"],
        code_style="clean"
    )

    # Start the agent
    await dev_agent.start()

    # Write code
    code_result = await dev_agent.write_code(
        description="Create a simple React component that displays cryptocurrency prices",
        language="javascript",
        framework="react",
        requirements=[
            "Fetch data from CoinGecko API",
            "Display prices for BTC, ETH, and SOL",
            "Update prices every 30 seconds",
            "Include error handling"
        ]
    )
    print("Code result:", code_result)

    # Review code
    review_result = await dev_agent.review_code(
        content="/* code to review */",
        language="javascript",
        framework="react"
    )
    print("Review result:", review_result)

    # Get code history
    code_history = await dev_agent.get_code_history(limit=10)
    print("Code history:", code_history)

    # Stop the agent
    await dev_agent.stop()

NumPy Integration

The Python wrapper provides seamless integration with NumPy for scientific computing and optimization:

import asyncio
import numpy as np
from juliaos import JuliaOS
from juliaos.swarms import DifferentialEvolution

# Define objective function using NumPy
def rastrigin(x: np.ndarray) -> float:
    return 10 * len(x) + np.sum(x**2 - 10 * np.cos(2 * np.pi * x))

async def main():
    # Initialize JuliaOS using async context manager
    async with JuliaOS(host="localhost", port=8052) as juliaos:
        # Create algorithm instance
        de = DifferentialEvolution(juliaos.bridge)

        # Define bounds as NumPy array
        bounds = np.array([[-5.12, 5.12]] * 5)  # 5-dimensional problem

        # Run optimization
        result = await de.optimize(
            objective_function=rastrigin,
            bounds=bounds,
            config={
                "population_size": 30,
                "max_generations": 100
            }
        )

        # Access result with NumPy arrays
        best_position = result['best_position_np']  # NumPy array
        best_fitness = result['best_fitness']

        print(f"Best position: {best_position}")
        print(f"Best fitness: {best_fitness}")

if __name__ == "__main__":
    asyncio.run(main())

The NumPy integration provides the following utilities:

• numpy_objective_wrapper: Wraps a NumPy-based objective function to make it compatible with JuliaOS swarm algorithms

• numpy_bounds_converter: Converts NumPy-style bounds to JuliaOS-compatible bounds

• numpy_result_converter: Converts JuliaOS optimization results to include NumPy arrays

LangChain Integration

The Python wrapper includes integration with LangChain:

import asyncio
import os
from dotenv import load_dotenv

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor
from langchain.chains import LLMChain
from langchain.prompts import ChatPromptTemplate

from juliaos import JuliaOS
from juliaos.agents import AgentType
from juliaos.langchain import (
    JuliaOSTradingAgentAdapter,
    SwarmOptimizationTool,
    BlockchainQueryTool
)

async def langchain_example():
    # Load environment variables
    load_dotenv()

    # Initialize JuliaOS using async context manager
    async with JuliaOS(host="localhost", port=8052) as juliaos_client:
        # Create a trading agent
        agent = await juliaos_client.agents.create_agent(
            name="LangChainTradingAgent",
            agent_type=AgentType.TRADING,
            config={
                "parameters": {
                    "risk_tolerance": 0.5,
                    "max_position_size": 1000.0
                }
            }
        )

        # Start the agent
        await agent.start()

        # Create a LangChain adapter for the agent
        llm = ChatOpenAI(temperature=0.7)
        agent_adapter = JuliaOSTradingAgentAdapter(agent)

        # Create tools
        tools = [
            SwarmOptimizationTool(bridge=juliaos_client.bridge),
            BlockchainQueryTool(bridge=juliaos_client.bridge)
        ]

        # Create a LangChain agent
        langchain_agent = agent_adapter.as_langchain_agent(
            llm=llm,
            tools=tools,
            verbose=True
        )

        # Run the agent
        result = await langchain_agent.arun(
            "Analyze the current market conditions for Ethereum and suggest a trading strategy."
        )

        print("LangChain agent result:", result)
    # Connection is automatically closed when exiting the context

Error Handling

The Python wrapper includes robust error handling:

import asyncio
from juliaos import JuliaOS
from juliaos.agents import AgentType
from juliaos.exceptions import JuliaOSError, ConnectionError, AgentError

async def error_handling_example():
    # Using try-except with async context manager
    try:
        async with JuliaOS(host="localhost", port=8052) as juliaos_client:
            try:
                agent = await juliaos_client.agents.create_agent(
                    name="MyAgent",
                    agent_type=AgentType.TRADING,
                    config={
                        "parameters": {
                            "risk_tolerance": 0.5,
                            "max_position_size": 1000.0
                        }
                    }
                )
            except AgentError as e:
                print("Failed to create agent:", e)
            except JuliaOSError as e:
                print("JuliaOS error:", e)
    except ConnectionError as e:
        print("Failed to connect to JuliaOS backend:", e)

# Alternative approach with manual connection management
async def alternative_error_handling():
    juliaos_client = JuliaOS(host="localhost", port=8052)

    try:
        await juliaos_client.connect()
    except ConnectionError as e:
        print("Failed to connect to JuliaOS backend:", e)
        return

    try:
        agent = await juliaos_client.agents.create_agent(
            name="MyAgent",
            agent_type=AgentType.TRADING,
            config={
                "parameters": {
                    "risk_tolerance": 0.5,
                    "max_position_size": 1000.0
                }
            }
        )
    except AgentError as e:
        print("Failed to create agent:", e)
    except JuliaOSError as e:
        print("JuliaOS error:", e)
    finally:
        await juliaos_client.disconnect()

See Also

• LangChain Integration - Learn more about integrating JuliaOS with LangChain

• Agents - Learn more about agents in JuliaOS

• Swarms - Learn more about swarms in JuliaOS

• API Reference - Detailed API reference for the Python wrapper

• NumPy Documentation - Official NumPy documentation