Skip to content

CodeLoop

pycodeloop.core.codeloop.CodeLoop

Import

You can import the CodeLoop class directly from pycodeloop:

from pycodeloop import CodeLoop, Config
from pycodeloop.providers import GenericProvider
Example

class pycodeloop.core.codeloop.CodeLoop

config = Config(
    provider=GenericProvider.from_json("path/to/config.json")
)

flow = CodeLoop(config=config)
flow.run("list the files in this repo and summarize the project")

Parameters:

Name Type Description Default
config Optional[Config]

Configuration class.

 None

Attributes:

Name Type Description
config Config
agent Agent
session Session

Conversation history, kept across run() calls so the agent remembers earlier turns.
Source code in pycodeloop/core/codeloop.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
class CodeLoop:
    """
    Import:
        You can import the **CodeLoop** class directly from pycodeloop:

            from pycodeloop import CodeLoop, Config
            from pycodeloop.providers import GenericProvider

    Example:
        `class` pycodeloop.core.codeloop.CodeLoop

            config = Config(
                provider=GenericProvider.from_json("path/to/config.json")
            )

            flow = CodeLoop(config=config)
            flow.run("list the files in this repo and summarize the project")

    Args:
        config (Optional[Config]): Configuration class.

    Attributes:
        config (Config):

        agent (Agent):

        session (Session): Conversation history, kept across `run()` calls
            so the agent remembers earlier turns.
    """

    def __init__(self, config: Config | None = None) -> None:
        self.config = config if config else Config()

        agent_kwargs = {}
        if self.config.system_prompt is not None:
            agent_kwargs["system_prompt"] = self.config.system_prompt

        self.agent = Agent(
            provider=self.config.provider,
            tools=self.config.tools,
            max_turns=self.config.max_turns,
            max_history_turns=self.config.max_history_turns,
            **agent_kwargs,
        )

        self.session = Session(
            system_prompt=self.agent.system_prompt, cwd=os.getcwd()
        )
        self._tracers: dict[str, JsonlTracer] = {}

    def run(
        self,
        prompt: str,
        session_key: str | None = None,
        images: list[str] | None = None,
        cancel_event: threading.Event | None = None,
    ) -> str:
        """Run prompt to completion, keeping history across calls.

        Pass `session_key` with a `Config(storage=...)` configured to
        resume a previous conversation: the session is loaded from
        storage before the run (falling back to the in-memory session
        on a cache miss) and saved back after.
        """
        can_persist = (
            session_key is not None and self.config.storage is not None
        )
        if can_persist:
            stored = self.config.storage.get(session_key)
            self.session = stored or Session(
                system_prompt=self.agent.system_prompt, cwd=os.getcwd()
            )
            self.agent.on_message = lambda: self.config.storage.post(
                session_key, self.session
            )
        else:
            self.agent.on_message = None

        self.agent.on_trace_event = (
            self._tracer_for(session_key or "global")
            if self.config.trace
            else None
        )

        usage_before = self.agent.usage
        with session_scope(session_key or "global"):
            result = self.agent.run(
                prompt,
                session=self.session,
                images=images,
                cancel_event=cancel_event,
            )
        usage_after = self.agent.usage

        _usage_tracker.record_usage(
            session_key or "global",
            usage_after.input_tokens - usage_before.input_tokens,
            usage_after.output_tokens - usage_before.output_tokens,
        )

        if can_persist:
            self.config.storage.post(session_key, self.session)

        return result

    def _tracer_for(self, key: str) -> JsonlTracer:
        tracer = self._tracers.get(key)
        if tracer is None:
            tracer = JsonlTracer(key)
            self._tracers[key] = tracer
        return tracer

    def ask(self, question: str) -> str:
        """One-shot Q&A against a read-only snapshot of the current
        session — doesn't touch `self.session`, run tools, or persist,
        so it's safe to call while `run()` is mid-turn on another thread."""
        messages = list(self.session.messages)
        last = messages[-1] if messages else None
        if last is not None and last.role == "assistant" and last.tool_calls:
            for call in last.tool_calls:
                messages.append(
                    Message(
                        role="tool",
                        tool_call_id=call["id"],
                        content="(pending — this tool hasn't finished yet)",
                    )
                )
        messages.append(Message(role="user", content=question))

        response = self.agent.provider.complete(
            system_prompt=self.agent.system_prompt,
            messages=messages,
            tools=[],
        )
        return response.text

agent = Agent(provider=(self.config.provider), tools=(self.config.tools), max_turns=(self.config.max_turns), max_history_turns=(self.config.max_history_turns), **agent_kwargs) instance-attribute

config = config if config else Config() instance-attribute

session = Session(system_prompt=(self.agent.system_prompt), cwd=(os.getcwd())) instance-attribute

__init__(config=None)

Source code in pycodeloop/core/codeloop.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def __init__(self, config: Config | None = None) -> None:
    self.config = config if config else Config()

    agent_kwargs = {}
    if self.config.system_prompt is not None:
        agent_kwargs["system_prompt"] = self.config.system_prompt

    self.agent = Agent(
        provider=self.config.provider,
        tools=self.config.tools,
        max_turns=self.config.max_turns,
        max_history_turns=self.config.max_history_turns,
        **agent_kwargs,
    )

    self.session = Session(
        system_prompt=self.agent.system_prompt, cwd=os.getcwd()
    )
    self._tracers: dict[str, JsonlTracer] = {}

ask(question)

One-shot Q&A against a read-only snapshot of the current session — doesn't touch self.session, run tools, or persist, so it's safe to call while run() is mid-turn on another thread.

Source code in pycodeloop/core/codeloop.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def ask(self, question: str) -> str:
    """One-shot Q&A against a read-only snapshot of the current
    session — doesn't touch `self.session`, run tools, or persist,
    so it's safe to call while `run()` is mid-turn on another thread."""
    messages = list(self.session.messages)
    last = messages[-1] if messages else None
    if last is not None and last.role == "assistant" and last.tool_calls:
        for call in last.tool_calls:
            messages.append(
                Message(
                    role="tool",
                    tool_call_id=call["id"],
                    content="(pending — this tool hasn't finished yet)",
                )
            )
    messages.append(Message(role="user", content=question))

    response = self.agent.provider.complete(
        system_prompt=self.agent.system_prompt,
        messages=messages,
        tools=[],
    )
    return response.text

run(prompt, session_key=None, images=None, cancel_event=None)

Run prompt to completion, keeping history across calls.

Pass session_key with a Config(storage=...) configured to resume a previous conversation: the session is loaded from storage before the run (falling back to the in-memory session on a cache miss) and saved back after.

Source code in pycodeloop/core/codeloop.py 
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def run(
    self,
    prompt: str,
    session_key: str | None = None,
    images: list[str] | None = None,
    cancel_event: threading.Event | None = None,
) -> str:
    """Run prompt to completion, keeping history across calls.

    Pass `session_key` with a `Config(storage=...)` configured to
    resume a previous conversation: the session is loaded from
    storage before the run (falling back to the in-memory session
    on a cache miss) and saved back after.
    """
    can_persist = (
        session_key is not None and self.config.storage is not None
    )
    if can_persist:
        stored = self.config.storage.get(session_key)
        self.session = stored or Session(
            system_prompt=self.agent.system_prompt, cwd=os.getcwd()
        )
        self.agent.on_message = lambda: self.config.storage.post(
            session_key, self.session
        )
    else:
        self.agent.on_message = None

    self.agent.on_trace_event = (
        self._tracer_for(session_key or "global")
        if self.config.trace
        else None
    )

    usage_before = self.agent.usage
    with session_scope(session_key or "global"):
        result = self.agent.run(
            prompt,
            session=self.session,
            images=images,
            cancel_event=cancel_event,
        )
    usage_after = self.agent.usage

    _usage_tracker.record_usage(
        session_key or "global",
        usage_after.input_tokens - usage_before.input_tokens,
        usage_after.output_tokens - usage_before.output_tokens,
    )

    if can_persist:
        self.config.storage.post(session_key, self.session)

    return result