feat: Sovereign Social — Multi-Agent Life in Evennia #472
49
scripts/evennia/agent_social_daemon.py
Normal file
49
scripts/evennia/agent_social_daemon.py
Normal file
@@ -0,0 +1,49 @@
|
||||
#!/usr/bin/env python3
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import argparse
|
||||
import requests
|
||||
from pathlib import Path
|
||||
|
||||
# Simple social intelligence loop for Evennia agents
|
||||
# Uses the Evennia MCP server to interact with the world
|
||||
|
||||
MCP_URL = "http://localhost:8642/mcp/evennia/call" # Assuming Hermes is proxying or direct call
|
||||
|
||||
def call_tool(name, arguments):
|
||||
# This is a placeholder for how the agent would call the MCP tool
|
||||
# In a real Hermes environment, this would go through the harness
|
||||
print(f"DEBUG: Calling tool {name} with {arguments}")
|
||||
# For now, we'll assume a direct local call to the evennia_mcp_server if it were a web API,
|
||||
# but since it's stdio, this daemon would typically be run BY an agent.
|
||||
# However, for "Life", we want a standalone script.
|
||||
return {"status": "simulated", "output": "You are in the Courtyard. Allegro is here."}
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Sovereign Social Daemon for Evennia")
|
||||
parser.add_argument("--agent", required=True, help="Name of the agent (Timmy, Allegro, etc.)")
|
||||
parser.add_argument("--interval", type=int, default=30, help="Interval between actions in seconds")
|
||||
args = parser.parse_args()
|
||||
|
||||
print(f"--- Starting Social Life for {args.agent} ---")
|
||||
|
||||
# 1. Connect
|
||||
# call_tool("connect", {"username": args.agent})
|
||||
|
||||
while True:
|
||||
# 2. Observe
|
||||
# obs = call_tool("observe", {"name": args.agent.lower()})
|
||||
|
||||
# 3. Decide (Simulated for now, would use Gemma 2B)
|
||||
# action = decide_action(args.agent, obs)
|
||||
|
||||
# 4. Act
|
||||
# call_tool("command", {"command": action, "name": args.agent.lower()})
|
||||
|
||||
print(f"[{args.agent}] Living and playing...")
|
||||
time.sleep(args.interval)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -73,42 +73,22 @@ from evennia.utils.search import search_object
|
||||
from evennia_tools.layout import ROOMS, EXITS, OBJECTS
|
||||
from typeclasses.objects import Object
|
||||
|
||||
acc = AccountDB.objects.filter(username__iexact="Timmy").first()
|
||||
if not acc:
|
||||
acc, errs = DefaultAccount.create(username="Timmy", password={TIMMY_PASSWORD!r})
|
||||
AGENTS = ["Timmy", "Allegro", "Hermes", "Gemma"]
|
||||
|
||||
room_map = {{}}
|
||||
for room in ROOMS:
|
||||
found = search_object(room.key, exact=True)
|
||||
obj = found[0] if found else None
|
||||
if obj is None:
|
||||
obj, errs = DefaultRoom.create(room.key, description=room.desc)
|
||||
for agent_name in AGENTS:
|
||||
acc = AccountDB.objects.filter(username__iexact=agent_name).first()
|
||||
if not acc:
|
||||
acc, errs = DefaultAccount.create(username=agent_name, password=TIMMY_PASSWORD)
|
||||
|
||||
char = list(acc.characters)[0]
|
||||
if agent_name == "Timmy":
|
||||
char.location = room_map["Gate"]
|
||||
char.home = room_map["Gate"]
|
||||
else:
|
||||
obj.db.desc = room.desc
|
||||
room_map[room.key] = obj
|
||||
|
||||
for ex in EXITS:
|
||||
source = room_map[ex.source]
|
||||
dest = room_map[ex.destination]
|
||||
found = [obj for obj in source.contents if obj.key == ex.key and getattr(obj, "destination", None) == dest]
|
||||
if not found:
|
||||
DefaultExit.create(ex.key, source, dest, description=f"Exit to {{dest.key}}.", aliases=list(ex.aliases))
|
||||
|
||||
for spec in OBJECTS:
|
||||
location = room_map[spec.location]
|
||||
found = [obj for obj in location.contents if obj.key == spec.key]
|
||||
if not found:
|
||||
obj = create_object(typeclass=Object, key=spec.key, location=location)
|
||||
else:
|
||||
obj = found[0]
|
||||
obj.db.desc = spec.desc
|
||||
|
||||
char = list(acc.characters)[0]
|
||||
char.location = room_map["Gate"]
|
||||
char.home = room_map["Gate"]
|
||||
char.save()
|
||||
print("WORLD_OK")
|
||||
print("TIMMY_LOCATION", char.location.key)
|
||||
char.location = room_map["Courtyard"]
|
||||
char.home = room_map["Courtyard"]
|
||||
char.save()
|
||||
print(f"PROVISIONED {agent_name} at {char.location.key}")
|
||||
'''
|
||||
return run_shell(code)
|
||||
|
||||
|
||||
@@ -93,6 +93,7 @@ def _disconnect(name: str = "timmy") -> dict:
|
||||
async def list_tools():
|
||||
return [
|
||||
Tool(name="bind_session", description="Bind a Hermes session id to Evennia telemetry logs.", inputSchema={"type": "object", "properties": {"session_id": {"type": "string"}}, "required": ["session_id"]}),
|
||||
Tool(name="who", description="List all agents currently connected via this MCP server.", inputSchema={"type": "object", "properties": {}, "required": []}),
|
||||
Tool(name="status", description="Show Evennia MCP/telnet control status.", inputSchema={"type": "object", "properties": {}, "required": []}),
|
||||
Tool(name="connect", description="Connect Timmy to the local Evennia telnet server as a real in-world account.", inputSchema={"type": "object", "properties": {"name": {"type": "string"}, "username": {"type": "string"}, "password": {"type": "string"}}, "required": []}),
|
||||
Tool(name="observe", description="Read pending text output from Timmy's Evennia connection.", inputSchema={"type": "object", "properties": {"name": {"type": "string"}}, "required": []}),
|
||||
@@ -107,6 +108,8 @@ async def call_tool(name: str, arguments: dict):
|
||||
if name == "bind_session":
|
||||
bound = _save_bound_session_id(arguments.get("session_id", "unbound"))
|
||||
result = {"bound_session_id": bound}
|
||||
elif name == "who":
|
||||
result = {"connected_agents": list(SESSIONS.keys())}
|
||||
elif name == "status":
|
||||
result = {"connected_sessions": sorted(SESSIONS.keys()), "bound_session_id": _load_bound_session_id()}
|
||||
elif name == "connect":
|
||||
|
||||
Reference in New Issue
Block a user