feat: Sovereign Social — Multi-Agent Life in Evennia #472

Merged
allegro merged 3 commits from feat/sovereign-social-evennia into main 2026-04-06 14:00:11 +00:00
3 changed files with 66 additions and 34 deletions

View 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()

View File

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

View File

@@ -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":