Python SDK

pip install agentfs-sdk

import asyncio
from agentfs_sdk import AgentFS, AgentFSOptions

async def main():
    # Open an agent filesystem
    agent = await AgentFS.open(AgentFSOptions(id='my-agent'))

    # Use key-value store
    await agent.kv.set('config', {'debug': True})
    config = await agent.kv.get('config')

    # Use filesystem
    await agent.fs.write_file('/notes.txt', 'Hello, AgentFS!')
    content = await agent.fs.read_file('/notes.txt')

    # Track tool calls
    call_id = await agent.tools.start('search', {'query': 'Python'})
    await agent.tools.success(call_id, {'results': ['result1']})

    await agent.close()

asyncio.run(main())

agent = await AgentFS.open(AgentFSOptions(id='my-agent'))

agent = await AgentFS.open(AgentFSOptions(path='./data/mydb.db'))

async with await AgentFS.open(AgentFSOptions(id='my-agent')) as agent:
    await agent.kv.set('key', 'value')
    # Database is closed when exiting

await agent.kv.set('user:123', {'name': 'Alice', 'age': 30})
await agent.kv.set('counter', 42)
await agent.kv.set('active', True)

user = await agent.kv.get('user:123')
# Returns None if key doesn't exist

# List all keys with prefix
keys = await agent.kv.list('user:')
# Returns: ['user:123', 'user:456', ...]

await agent.kv.delete('user:123')

await agent.fs.write_file('/data/config.json', '{"key": "value"}')

# Write bytes
await agent.fs.write_file('/data/image.png', image_bytes)

# Read as string (default)
content = await agent.fs.read_file('/data/config.json')

# Read as bytes
data = await agent.fs.read_file('/data/image.png', encoding=None)

entries = await agent.fs.readdir('/data')
# Returns: ['config.json', 'image.png', ...]

stats = await agent.fs.stat('/data/config.json')
print(f"Size: {stats.size} bytes")
print(f"Modified: {stats.mtime}")
print(f"Is file: {stats.is_file()}")
print(f"Is directory: {stats.is_dir()}")

await agent.fs.delete_file('/data/config.json')

await agent.fs.mkdir('/data/subdir')

# Start a tool call
call_id = await agent.tools.start('search', {'query': 'Python'})

# ... perform the operation ...

# Mark as successful
await agent.tools.success(call_id, {'results': ['result1', 'result2']})

# Or mark as failed
await agent.tools.error(call_id, 'Connection timeout')

await agent.tools.record(
    name='search',
    started_at=1234567890.0,
    completed_at=1234567892.0,
    parameters={'query': 'Python'},
    result={'results': ['result1']}
)

# Get by name
calls = await agent.tools.get_by_name('search', limit=10)

# Get recent calls
recent = await agent.tools.get_recent(since=1234567890.0)

# Get by ID
call = await agent.tools.get(42)

stats = await agent.tools.get_stats()
for stat in stats:
    print(f"{stat.name}:")
    print(f"  Total calls: {stat.total_calls}")
    print(f"  Successful: {stat.successful}")
    print(f"  Failed: {stat.failed}")
    print(f"  Avg duration: {stat.avg_duration_ms:.2f}ms")

import asyncio
import time
from agentfs_sdk import AgentFS, AgentFSOptions

async def main():
    async with await AgentFS.open(AgentFSOptions(id='demo-agent')) as agent:
        # Store configuration
        await agent.kv.set('agent:config', {
            'model': 'gpt-4',
            'temperature': 0.7
        })

        # Create a research document
        research = """
        # Research Notes

        ## Topic: AgentFS

        AgentFS provides persistent storage for AI agents.
        """

        await agent.fs.write_file('/research/agentfs.md', research)

        # Track a simulated tool call
        call_id = await agent.tools.start('web_search', {
            'query': 'AgentFS documentation'
        })

        # Simulate some work
        await asyncio.sleep(0.1)

        await agent.tools.success(call_id, {
            'results_count': 5,
            'top_result': 'https://docs.turso.tech/agentfs'
        })

        # List what we created
        files = await agent.fs.readdir('/research')
        print(f"Files: {files}")

        # Get tool stats
        stats = await agent.tools.get_stats()
        for stat in stats:
            print(f"{stat.name}: {stat.total_calls} calls")

asyncio.run(main())