Skip to main content

A graph-based task management framework for AI agents

Project description

SocialSeed Tasker

๐Ÿ”ญ Graph-Native Engineering & Autonomous Agent Governance

A specialized framework that leverages Neo4j to provide AI agents with infinite architectural context and strict governance.

Python 3.10+ Hexagonal Architecture Neo4j Only GraphRAG License: Apache 2.0 PRs Welcome


๐Ÿš€ Quick Start for AI Agents

Starting the Server

# Using Docker Compose (recommended - starts API, Frontend, and Neo4j)
docker compose up -d

# Or start only the API directly
pip install socialseed-tasker
uvicorn socialseed_tasker.entrypoints.web_api.api:app --host 0.0.0.0 --port 8000

Services Available

Service URL Description
Neo4j Browser http://localhost:7474 Graph database UI (neo4j/neoSocial)
REST API http://localhost:8000 For AI agents to manage issues
Frontend http://localhost:8080 Human UI (Kanban board)
API Docs http://localhost:8000/docs OpenAPI documentation

๐Ÿ”Œ REST API Reference for AI Agents

Base URL

http://localhost:8000/api/v1

Authentication

Currently no authentication required. Set ALLOWED_ORIGINS env var in production.


Components

Components represent different parts of your project (services, modules, packages).

Create Component

curl -X POST http://localhost:8000/api/v1/components \
  -H "Content-Type: application/json" \
  -d '{
    "name": "auth-service",
    "description": "Authentication microservice",
    "project": "social-network"
  }'

List Components

curl http://localhost:8000/api/v1/components

Issues

Create Issue

curl -X POST http://localhost:8000/api/v1/issues \
  -H "Content-Type: application/json" \
  -d '{
    "title": "Fix login bug with special characters",
    "description": "Users cannot login when password contains special chars",
    "priority": "HIGH",
    "component_id": "<component-uuid>",
    "labels": ["bug", "security"]
  }'

Priority values: LOW, MEDIUM, HIGH, CRITICAL

List Issues

# All issues
curl http://localhost:8000/api/v1/issues

# Filter by status
curl "http://localhost:8000/api/v1/issues?status=OPEN"

# Filter by component
curl "http://localhost:8000/api/v1/issues?component=<component-id>"

Update Issue

# Update status
curl -X PATCH http://localhost:8000/api/v1/issues/<issue-id> \
  -H "Content-Type: application/json" \
  -d '{"status": "IN_PROGRESS"}'

# Mark that an AI agent is working on this issue
curl -X PATCH http://localhost:8000/api/v1/issues/<issue-id> \
  -H "Content-Type: application/json" \
  -d '{"agent_working": true}'

# Update priority
curl -X PATCH http://localhost:8000/api/v1/issues/<issue-id> \
  -H "Content-Type: application/json" \
  -d '{"priority": "CRITICAL"}'

# Update description
curl -X PATCH http://localhost:8000/api/v1/issues/<issue-id> \
  -H "Content-Type: application/json" \
  -d '{"description": "Updated description"}'

# Close an issue
curl -X POST http://localhost:8000/api/v1/issues/<issue-id>/close

Status values: OPEN, IN_PROGRESS, BLOCKED, CLOSED

Note: Currently, issue show and update endpoints require the full UUID. Short ID support is planned for v0.6.0.

Delete Issue

curl -X DELETE http://localhost:8000/api/v1/issues/<issue-id>

Dependencies

Dependencies define which issues block others. AI agents use this to understand what can be worked on.

Add Dependency

# Issue A depends on Issue B (B must be completed first)
curl -X POST http://localhost:8000/api/v1/issues/<issue-a-id>/dependencies \
  -H "Content-Type: application/json" \
  -d '{"depends_on_id": "<issue-b-id>"}'

List Dependencies

# What does this issue depend on?
curl http://localhost:8000/api/v1/issues/<issue-id>/dependencies

Remove Dependency

curl -X DELETE http://localhost:8000/api/v1/issues/<issue-a-id>/dependencies/<issue-b-id>

Agent Working Indicator

AI agents can set agent_working: true on an issue to signal they're actively working on it. This displays a cyan robot icon on the Kanban board.

import requests

# Tell the system you're working on this issue
requests.patch(
    "http://localhost:8000/api/v1/issues/<issue-id>",
    json={"agent_working": True}
)

# When done, clear the flag
requests.patch(
    "http://localhost:8000/api/v1/issues/<issue-id>",
    json={"agent_working": False}
)

๐Ÿค– AI Agent Workflow

Recommended Workflow for AI Agents

import requests
from datetime import datetime

API_BASE = "http://localhost:8000/api/v1"

def start_working_on_issue(issue_id, todo_items):
    """AI agent starts working on an issue - updates status and sets todo."""
    
    # 1. Create a detailed TODO list in the description
    todo_text = "## TODO:\n" + "\n".join([f"- [ ] {item}" for item in todo_items])
    todo_text += f"\n\n## Progress (started {datetime.now().strftime('%Y-%m-%d %H:%M')}):\n"
    
    requests.patch(f"{API_BASE}/issues/{issue_id}", json={
        "description": todo_text,
        "agent_working": True,
        "status": "IN_PROGRESS"
    })

def update_progress(issue_id, completed_item, next_step):
    """Update progress on the issue."""
    
    # Get current description
    issue = requests.get(f"{API_BASE}/issues/{issue_id}").json()["data"]
    desc = issue.get("description", "")
    
    # Mark completed item
    desc = desc.replace(f"- [ ] {completed_item}", f"- [x] {completed_item}")
    
    # Add progress note
    desc += f"\n- **In progress**: {next_step}"
    
    requests.patch(f"{API_BASE}/issues/{issue_id}", json={
        "description": desc
    })

def finish_issue(issue_id, solution_summary):
    """Mark issue as completed with solution summary."""
    
    # Get current description
    issue = requests.get(f"{API_BASE}/issues/{issue_id}").json()["data"]
    desc = issue.get("description", "")
    
    # Add solution summary
    desc += f"\n\n## Solution:\n{solution_summary}"
    
    # Close the issue
    requests.post(f"{API_BASE}/issues/{issue_id}/close")
    
    # Clear agent working flag
    requests.patch(f"{API_BASE}/issues/{issue_id}", json={
        "description": desc,
        "agent_working": False
    })

def create_issue_and_work(title, component_id, description=""): """Create an issue and start working on it."""

# 1. Create the issue
response = requests.post(f"{API_BASE}/issues", json={
    "title": title,
    "description": description,
    "component_id": component_id,
    "priority": "MEDIUM"
})
issue = response.json()["data"]
issue_id = issue["id"]

# 2. Mark as agent working (shows on board)
requests.patch(f"{API_BASE}/{issue_id}", json={"agent_working": True})

# 3. Move to in progress
requests.patch(f"{API_BASE}/{issue_id}", json={"status": "IN_PROGRESS"})

# 4. Do the work...

# 5. Mark as done
requests.post(f"{API_BASE}/{issue_id}/close")

# 6. Clear agent working flag
requests.patch(f"{API_BASE}/{issue_id}", json={"agent_working": False})

return issue_id

### Full Example: AI Agent Solving an Issue

```python
import requests

API_BASE = "http://localhost:8000/api/v1"

def solve_issue(issue_id, problem_description):
    """
    AI agent solves an issue, keeping the board updated with progress.
    """
    
    # Step 1: Analyze and plan
    todo_items = [
        "Analyze the problem and identify root cause",
        "Write test to reproduce the issue",
        "Implement the fix",
        "Run tests to verify the solution",
        "Update documentation if needed"
    ]
    
    # Step 2: Start working - update status and add TODO to description
    initial_desc = f"## Problem\n{problem_description}\n\n"
    initial_desc += "## TODO:\n" + "\n".join([f"- [ ] {item}" for item in todo_items])
    initial_desc += f"\n\n## Started at: {datetime.now().isoformat()}"
    
    requests.patch(f"{API_BASE}/issues/{issue_id}", json={
        "description": initial_desc,
        "status": "IN_PROGRESS",
        "agent_working": True
    })
    
    # Step 3: First TODO complete - Analyze
    update_issue_description(issue_id, todo_items[0], "Analyzing root cause...")
    
    # Step 4: Found the issue - update with findings
    update_issue_description(issue_id, todo_items[0], 
        "Root cause: Missing null check in auth handler")
    
    # Step 5: Write test
    update_issue_description(issue_id, todo_items[1], 
        "Test written: test_login_with_special_chars")
    
    # Step 6: Implement fix
    update_issue_description(issue_id, todo_items[2], 
        "Fix implemented: Added null validation")
    
    # Step 7: Tests pass
    update_issue_description(issue_id, todo_items[3], 
        "All 45 tests pass")
    
    # Step 8: Close issue with summary
    solution_summary = """
## Solution Applied
- Added null validation for password field
- Added test case with special characters: !@#$%^&*()
- All existing tests continue to pass

## Files Changed
- src/auth/handlers.py (line 45-47)
- tests/test_auth.py (added test_login_with_special_chars)
"""
    
    requests.post(f"{API_BASE}/issues/{issue_id}/close")
    requests.patch(f"{API_BASE}/issues/{issue_id}", json={
        "description": initial_desc + solution_summary,
        "agent_working": False
    })


def update_issue_description(issue_id, completed_item, next_action):
    """Helper to mark a TODO complete and note next action."""
    
    # Get current description
    issue = requests.get(f"{API_BASE}/issues/{issue_id}").json()["data"]
    desc = issue.get("description", "")
    
    # Mark completed item
    desc = desc.replace(f"- [ ] {completed_item}", f"- [x] {completed_item}")
    
    # Add progress note with timestamp
    timestamp = datetime.now().strftime("%H:%M")
    desc += f"\n[{timestamp}] **Next**: {next_action}"
    
    requests.patch(f"{API_BASE}/issues/{issue_id}", json={
        "description": desc
    })

Why Keep the Board Updated?

Reason Description
Transparency Humans can see what the AI is doing
Traceability Full history of the solution is in the issue
Collaboration Other agents or humans can see progress
Debugging If something goes wrong, the trail is clear
State Sync The Kanban board always reflects reality

def get_workable_issues(): """Get issues that can be worked on (not blocked).""" all_issues = requests.get(f"{API_BASE}/issues").json()["data"]["items"]

workable = []
for issue in all_issues:
    if issue["status"] == "CLOSED":
        continue
    
    # Check dependencies - if all dependencies are closed, it's workable
    deps = requests.get(f"{API_BASE}/issues/{issue['id']}/dependencies").json()["data"]
    
    if not deps or all(d["status"] == "CLOSED" for d in deps):
        workable.append(issue)

return workable

---

## ๐Ÿ“‹ Complete Example: Full Workflow

```bash
# 1. Create a component
COMPONENT=$(curl -s -X POST http://localhost:8000/api/v1/components \
  -H "Content-Type: application/json" \
  -d '{"name": "api-service", "project": "myapp"}' | jq -r '.data.id')

echo "Created component: $COMPONENT"

# 2. Create multiple issues
ISSUE1=$(curl -s -X POST http://localhost:8000/api/v1/issues \
  -H "Content-Type: application/json" \
  -d "{\"title\": \"Setup database schema\", \"component_id\": \"$COMPONENT\", \"priority\": \"HIGH\"}" | jq -r '.data.id')

ISSUE2=$(curl -s -X POST http://localhost:8000/api/v1/issues \
  -H "Content-Type: application/json" \
  -d "{\"title\": \"Create API endpoints\", \"component_id\": \"$COMPONENT\", \"priority\": \"HIGH\"}" | jq -r '.data.id')

# 3. Add dependency: API endpoints depend on database schema
curl -X POST "http://localhost:8000/api/v1/issues/$ISSUE2/dependencies" \
  -H "Content-Type: application/json" \
  -d "{\"depends_on_id\": \"$ISSUE1\"}"

# 4. Mark AI is working on database schema
curl -X PATCH "http://localhost:8000/api/v1/issues/$ISSUE1" \
  -H "Content-Type: application/json" \
  -d '{"agent_working": true, "status": "IN_PROGRESS"}'

# 5. After completing database, close it and work on API
curl -X POST "http://localhost:8000/api/v1/issues/$ISSUE1/close"
curl -X PATCH "http://localhost:8000/api/v1/issues/$ISSUE2" \
  -H "Content-Type: application/json" \
  -d '{"agent_working": true, "status": "IN_PROGRESS"}'

# 6. Check all issues
curl http://localhost:8000/api/v1/issues | jq '.data.items[] | {title, status, agent_working, dependencies}'

๐Ÿ” Finding Workable Issues

AI agents can query to find issues that are ready to work on:

# Get all open issues
curl "http://localhost:8000/api/v1/issues?status=OPEN" | jq '.data.items[] | .title'

# For each, check if dependencies are resolved
curl "http://localhost:8000/api/v1/issues/<id>/dependencies"

๐Ÿง  GraphRAG: Infinite Context for AI Agents

The Tasker graph isn't just for tracking issuesโ€”it's a knowledge graph that provides AI agents with infinite context before proposing solutions.

How It Works

  1. Dependency Tracking: Every issue relationship is stored as a directed edge in Neo4j
  2. Impact Analysis: Before working on an issue, agents can query the full dependency chain to understand downstream effects
  3. Root Cause Discovery: Closed issues are linked to test failures, enabling automated root cause analysis

Example: Impact Analysis Before a Fix

# Before refactoring the Auth module, check what would be affected
import requests

issue_id = "<auth-module-issue-id>"
impact = requests.get(f"{API_BASE}/analyze/impact/{issue_id}").json()["data"]

print(f"Directly affected: {len(impact['directly_affected'])} issues")
print(f"Transitively affected: {len(impact['transitively_affected'])} issues")
print(f"Risk level: {impact['risk_level']}")

This enables the agent to make informed decisions based on the complete architectural context, not just the isolated issue.


๐Ÿ”ง Environment Variables

Variable Default Description
TASKER_NEO4J_URI bolt://localhost:7687 Neo4j connection URI
TASKER_NEO4J_USER neo4j Neo4j username
TASKER_NEO4J_PASSWORD (none) Neo4j password (required)
API_PORT 8000 API server port

Note: When using Docker Compose, credentials are set via NEO4J_PASSWORD in the compose file. Default is neo4j/neoSocial.


๐Ÿณ Docker Compose

The included docker-compose.yml starts:

  • Neo4j (port 7474/7687) - Graph database
  • API (port 8000) - REST API for AI agents
  • Frontend (port 8080) - Human Kanban board
# Start everything
docker compose up -d

# View logs
docker compose logs -f

# Stop everything
docker compose down

๐Ÿ“Š Architecture

โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”
โ”‚   AI Agent / Human UI    โ”‚
โ”‚   REST API (port 8000)   โ”‚
โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ฌโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜
             โ–ผ
โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”
โ”‚      Application Core        โ”‚
โ”‚  (Hexagonal Architecture)    โ”‚
โ”‚ โ€ข Governance Engine           โ”‚
โ”‚ โ€ข Dependency BFS Analysis     โ”‚
โ”‚ โ€ข Root Cause Detection        โ”‚
โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ฌโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜
             โ–ผ
โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”
โ”‚      Neo4j Graph DB          โ”‚
โ”‚ (The Source of Truth)        โ”‚
โ”‚ โ€ข Relationship Tracking       โ”‚
โ”‚ โ€ข Causal Traceability         โ”‚
โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜

๐Ÿ”— Related 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

socialseed_tasker-0.5.0.tar.gz (55.4 kB view details)

Uploaded Source

Built Distribution

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

socialseed_tasker-0.5.0-py3-none-any.whl (62.1 kB view details)

Uploaded Python 3

File details

Details for the file socialseed_tasker-0.5.0.tar.gz.

File metadata

  • Download URL: socialseed_tasker-0.5.0.tar.gz
  • Upload date:
  • Size: 55.4 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.1.0 CPython/3.13.12

File hashes

Hashes for socialseed_tasker-0.5.0.tar.gz
Algorithm Hash digest
SHA256 ddc6fb68a316a03c31f9e5e095ef8d2d915bb46b6610b8bb24adf27e45004f4c
MD5 f7da4f720bc2c6947836a68abe72ccd5
BLAKE2b-256 255225457ce341675b378b2bc5e18e0174d53f32df91f55888360666e307b97a

See more details on using hashes here.

File details

Details for the file socialseed_tasker-0.5.0-py3-none-any.whl.

File metadata

File hashes

Hashes for socialseed_tasker-0.5.0-py3-none-any.whl
Algorithm Hash digest
SHA256 1caa34272e2de2c47482cd946cc0930e46de984ec1b4837f48119374415ffa54
MD5 e1f3e7d872e083c25763360c4e226ecb
BLAKE2b-256 78114390fcca099de54c94724b5228352a5c7125af0009317b4c6f5279e0b3d2

See more details on using hashes here.

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