diff --git a/dev/integration/README.md b/dev/integration/README.md
new file mode 100644
index 00000000..614462c9
--- /dev/null
+++ b/dev/integration/README.md
@@ -0,0 +1,2 @@
+# Microsoft 365 Agents SDK for Python Integration Testing Framework
+
diff --git a/dev/integration/requirements.txt b/dev/integration/requirements.txt
new file mode 100644
index 00000000..d524e63a
--- /dev/null
+++ b/dev/integration/requirements.txt
@@ -0,0 +1 @@
+aioresponses
\ No newline at end of file
diff --git a/dev/integration/src/__init__.py b/dev/integration/src/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/dev/integration/src/core/__init__.py b/dev/integration/src/core/__init__.py
new file mode 100644
index 00000000..a8f5206c
--- /dev/null
+++ b/dev/integration/src/core/__init__.py
@@ -0,0 +1,21 @@
+from .application_runner import ApplicationRunner
+from .aiohttp import AiohttpEnvironment
+from .client import (
+    AgentClient,
+    ResponseClient,
+)
+from .environment import Environment
+from .integration import integration, IntegrationFixtures
+from .sample import Sample
+
+
+__all__ = [
+    "AgentClient",
+    "ApplicationRunner",
+    "AiohttpEnvironment",
+    "ResponseClient",
+    "Environment",
+    "integration",
+    "IntegrationFixtures",
+    "Sample",
+]
diff --git a/dev/integration/src/core/aiohttp/__init__.py b/dev/integration/src/core/aiohttp/__init__.py
new file mode 100644
index 00000000..82d2d1d0
--- /dev/null
+++ b/dev/integration/src/core/aiohttp/__init__.py
@@ -0,0 +1,4 @@
+from .aiohttp_environment import AiohttpEnvironment
+from .aiohttp_runner import AiohttpRunner
+
+__all__ = ["AiohttpEnvironment", "AiohttpRunner"]
diff --git a/dev/integration/src/core/aiohttp/aiohttp_environment.py b/dev/integration/src/core/aiohttp/aiohttp_environment.py
new file mode 100644
index 00000000..c8256618
--- /dev/null
+++ b/dev/integration/src/core/aiohttp/aiohttp_environment.py
@@ -0,0 +1,58 @@
+from tkinter import E
+from aiohttp.web import Request, Response, Application, run_app
+
+from microsoft_agents.hosting.aiohttp import (
+    CloudAdapter,
+    jwt_authorization_middleware,
+    start_agent_process,
+)
+from microsoft_agents.hosting.core import (
+    Authorization,
+    AgentApplication,
+    TurnState,
+    MemoryStorage,
+)
+from microsoft_agents.authentication.msal import MsalConnectionManager
+from microsoft_agents.activity import load_configuration_from_env
+
+from ..application_runner import ApplicationRunner
+from ..environment import Environment
+from .aiohttp_runner import AiohttpRunner
+
+
+class AiohttpEnvironment(Environment):
+    """An environment for aiohttp-hosted agents."""
+
+    async def init_env(self, environ_config: dict) -> None:
+        environ_config = environ_config or {}
+
+        self.config = load_configuration_from_env(environ_config)
+
+        self.storage = MemoryStorage()
+        self.connection_manager = MsalConnectionManager(**self.config)
+        self.adapter = CloudAdapter(connection_manager=self.connection_manager)
+        self.authorization = Authorization(
+            self.storage, self.connection_manager, **self.config
+        )
+
+        self.agent_application = AgentApplication[TurnState](
+            storage=self.storage,
+            adapter=self.adapter,
+            authorization=self.authorization,
+            **self.config
+        )
+
+    def create_runner(self, host: str, port: int) -> ApplicationRunner:
+
+        async def entry_point(req: Request) -> Response:
+            agent: AgentApplication = req.app["agent_app"]
+            adapter: CloudAdapter = req.app["adapter"]
+            return await start_agent_process(req, agent, adapter)
+
+        APP = Application(middlewares=[jwt_authorization_middleware])
+        APP.router.add_post("/api/messages", entry_point)
+        APP["agent_configuration"] = self.connection_manager.get_default_connection_configuration()
+        APP["agent_app"] = self.agent_application
+        APP["adapter"] = self.adapter
+
+        return AiohttpRunner(APP, host, port)
diff --git a/dev/integration/src/core/aiohttp/aiohttp_runner.py b/dev/integration/src/core/aiohttp/aiohttp_runner.py
new file mode 100644
index 00000000..3b6780d4
--- /dev/null
+++ b/dev/integration/src/core/aiohttp/aiohttp_runner.py
@@ -0,0 +1,115 @@
+from typing import Optional
+from typing import Optional
+from threading import Thread, Event
+import asyncio
+
+from aiohttp import ClientSession
+from aiohttp.web import Application, Request, Response
+from aiohttp.web_runner import AppRunner, TCPSite
+
+from ..application_runner import ApplicationRunner
+
+
+class AiohttpRunner(ApplicationRunner):
+    """A runner for aiohttp applications."""
+
+    def __init__(self, app: Application, host: str = "localhost", port: int = 8000):
+        assert isinstance(app, Application)
+        super().__init__(app)
+
+        url = f"{host}:{port}"
+        self._host = host
+        self._port = port
+        if "http" not in url:
+            url = f"http://{url}"
+        self._url = url
+
+        self._app.router.add_get("/shutdown", self._shutdown_route)
+
+        self._server_thread: Optional[Thread] = None
+        self._shutdown_event = Event()
+        self._runner: Optional[AppRunner] = None
+        self._site: Optional[TCPSite] = None
+
+    @property
+    def url(self) -> str:
+        return self._url
+
+    async def _start_server(self) -> None:
+        try:
+            assert isinstance(self._app, Application)
+
+            self._runner = AppRunner(self._app)
+            await self._runner.setup()
+            self._site = TCPSite(self._runner, self._host, self._port)
+            await self._site.start()
+
+            # Wait for shutdown signal
+            while not self._shutdown_event.is_set():
+                await asyncio.sleep(0.1)
+
+            # Cleanup
+            await self._site.stop()
+            await self._runner.cleanup()
+
+        except Exception as error:
+            raise error
+
+    async def __aenter__(self):
+        if self._server_thread:
+            raise RuntimeError("ResponseClient is already running.")
+
+        self._shutdown_event.clear()
+        self._server_thread = Thread(
+            target=lambda: asyncio.run(self._start_server()), daemon=True
+        )
+        self._server_thread.start()
+
+        # Wait a moment to ensure the server starts
+        await asyncio.sleep(0.5)
+
+        return self
+
+    async def _stop_server(self):
+        if not self._server_thread:
+            raise RuntimeError("ResponseClient is not running.")
+
+        try:
+            async with ClientSession() as session:
+                async with session.get(
+                    f"http://{self._host}:{self._port}/shutdown"
+                ) as response:
+                    pass  # Just trigger the shutdown
+        except Exception:
+            pass  # Ignore errors during shutdown request
+
+        # Set shutdown event as fallback
+        self._shutdown_event.set()
+
+        # Wait for the server thread to finish
+        self._server_thread.join(timeout=5.0)
+        self._server_thread = None
+
+    async def _shutdown_route(self, request: Request) -> Response:
+        """Handle shutdown request by setting the shutdown event"""
+        self._shutdown_event.set()
+        return Response(status=200, text="Shutdown initiated")
+
+    async def __aexit__(self, exc_type, exc, tb):
+        if not self._server_thread:
+            raise RuntimeError("ResponseClient is not running.")
+        try:
+            async with ClientSession() as session:
+                async with session.get(
+                    f"http://{self._host}:{self._port}/shutdown"
+                ) as response:
+                    pass  # Just trigger the shutdown
+        except Exception:
+            pass  # Ignore errors during shutdown request
+
+        # Set shutdown event as fallback
+        self._shutdown_event.set()
+
+        # Wait for the server thread to finish
+        self._server_thread.join(timeout=5.0)
+        self._server_thread = None
diff --git a/dev/integration/src/core/application_runner.py b/dev/integration/src/core/application_runner.py
new file mode 100644
index 00000000..ebbc56f9
--- /dev/null
+++ b/dev/integration/src/core/application_runner.py
@@ -0,0 +1,42 @@
+import asyncio
+from abc import ABC, abstractmethod
+from typing import Any, Optional
+from threading import Thread
+
+
+class ApplicationRunner(ABC):
+    """Base class for application runners."""
+
+    def __init__(self, app: Any):
+        self._app = app
+        self._thread: Optional[Thread] = None
+
+    @abstractmethod
+    async def _start_server(self) -> None:
+        raise NotImplementedError(
+            "Start server method must be implemented by subclasses"
+        )
+
+    async def _stop_server(self) -> None:
+        pass
+
+    async def __aenter__(self) -> None:
+
+        if self._thread:
+            raise RuntimeError("Server is already running")
+
+        def target():
+            asyncio.run(self._start_server())
+
+        self._thread = Thread(target=target, daemon=True)
+        self._thread.start()
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
+
+        if self._thread:
+            await self._stop_server()
+
+            self._thread.join()
+            self._thread = None
+        else:
+            raise RuntimeError("Server is not running")
diff --git a/dev/integration/src/core/client/__init__.py b/dev/integration/src/core/client/__init__.py
new file mode 100644
index 00000000..1d59411e
--- /dev/null
+++ b/dev/integration/src/core/client/__init__.py
@@ -0,0 +1,7 @@
+from .agent_client import AgentClient
+from .response_client import ResponseClient
+
+__all__ = [
+    "AgentClient",
+    "ResponseClient",
+]
diff --git a/dev/integration/src/core/client/agent_client.py b/dev/integration/src/core/client/agent_client.py
new file mode 100644
index 00000000..bbf650b6
--- /dev/null
+++ b/dev/integration/src/core/client/agent_client.py
@@ -0,0 +1,136 @@
+import os
+import json
+import asyncio
+from typing import Optional, cast
+
+from aiohttp import ClientSession
+from msal import ConfidentialClientApplication
+
+from microsoft_agents.activity import Activity, ActivityTypes, DeliveryModes, ConversationAccount
+
+from ..utils import _populate_incoming_activity
+
+
+class AgentClient:
+
+    def __init__(
+        self,
+        agent_url: str,
+        cid: str,
+        client_id: str,
+        tenant_id: str,
+        client_secret: str,
+        service_url: Optional[str] = None,
+        default_timeout: float = 5.0,
+    ):
+        self._agent_url = agent_url
+        self._cid = cid
+        self._client_id = client_id
+        self._tenant_id = tenant_id
+        self._client_secret = client_secret
+        self._service_url = service_url
+        self._headers = None
+        self._default_timeout = default_timeout
+
+        self._client: Optional[ClientSession] = None
+
+    @property
+    def agent_url(self) -> str:
+        return self._agent_url
+
+    @property
+    def service_url(self) -> Optional[str]:
+        return self._service_url
+
+    async def get_access_token(self) -> str:
+
+        msal_app = ConfidentialClientApplication(
+            client_id=self._client_id,
+            client_credential=self._client_secret,
+            authority=f"https://login.microsoftonline.com/{self._tenant_id}",
+        )
+
+        res = msal_app.acquire_token_for_client(scopes=[f"{self._client_id}/.default"])
+        token = res.get("access_token") if res else None
+        if not token:
+            raise Exception("Could not obtain access token")
+        return token
+
+    async def _init_client(self) -> None:
+        if not self._client:
+            if self._client_secret:
+                token = await self.get_access_token()
+                self._headers = {
+                    "Authorization": f"Bearer {token}",
+                    "Content-Type": "application/json",
+                }
+            else:
+                self._headers = {"Content-Type": "application/json"}
+
+            self._client = ClientSession(
+                base_url=self._agent_url, headers=self._headers
+            )
+
+    async def send_request(self, activity: Activity, sleep: float = 0) -> str:
+
+        await self._init_client()
+        assert self._client
+
+        if activity.conversation:
+            activity.conversation.id = self._cid
+        else:
+            activity.conversation = ConversationAccount(id=self._cid or "<conversation_id>")
+
+        if self.service_url:
+            activity.service_url = self.service_url
+
+        activity = _populate_incoming_activity(activity)
+
+        async with self._client.post(
+            "api/messages",
+            headers=self._headers,
+            json=activity.model_dump(by_alias=True, exclude_unset=True, exclude_none=True, mode="json"),
+        ) as response:
+            content = await response.text()
+            if not response.ok:
+                raise Exception(f"Failed to send activity: {response.status}")
+            await asyncio.sleep(sleep)
+            return content
+
+    def _to_activity(self, activity_or_text: Activity | str) -> Activity:
+        if isinstance(activity_or_text, str):
+            activity = Activity(
+                type=ActivityTypes.message,
+                text=activity_or_text,
+            )
+            return activity
+        else:
+            return cast(Activity, activity_or_text)
+
+    async def send_activity(
+        self, activity_or_text: Activity | str, sleep: float = 0, timeout: Optional[float] = None
+    ) -> str:
+        timeout = timeout or self._default_timeout
+        activity = self._to_activity(activity_or_text)
+        content = await self.send_request(activity, sleep=sleep)
+        return content
+
+    async def send_expect_replies(
+        self, activity_or_text: Activity | str, sleep: float = 0, timeout: Optional[float] = None
+    ) -> list[Activity]:
+        timeout = timeout or self._default_timeout
+        activity = self._to_activity(activity_or_text)
+        activity.delivery_mode = DeliveryModes.expect_replies
+        activity.service_url = activity.service_url or "http://localhost" # temporary fix
+
+        content = await self.send_request(activity, sleep=sleep)
+
+        activities_data = json.loads(content).get("activities", [])
+        activities = [Activity.model_validate(act) for act in activities_data]
+
+        return activities
+
+    async def close(self) -> None:
+        if self._client:
+            await self._client.close()
+            self._client = None
diff --git a/dev/integration/src/core/client/auto_client.py b/dev/integration/src/core/client/auto_client.py
new file mode 100644
index 00000000..dcea531b
--- /dev/null
+++ b/dev/integration/src/core/client/auto_client.py
@@ -0,0 +1,18 @@
+# from microsoft_agents.activity import Activity
+
+# from ..agent_client import AgentClient
+
+# class AutoClient:
+
+#     def __init__(self, agent_client: AgentClient):
+#         self._agent_client = agent_client
+
+#     async def generate_message(self) -> str:
+#         pass
+
+#     async def run(self, max_turns: int = 10, time_between_turns: float = 2.0) -> None:
+
+#         for i in range(max_turns):
+#             await self._agent_client.send_activity(
+#                 Activity(type="message", text=self.generate_message())
+#             )
diff --git a/dev/integration/src/core/client/response_client.py b/dev/integration/src/core/client/response_client.py
new file mode 100644
index 00000000..d93bfb80
--- /dev/null
+++ b/dev/integration/src/core/client/response_client.py
@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+import sys
+from io import StringIO
+from typing import Optional
+from threading import Lock, Thread, Event
+import asyncio
+
+from aiohttp.web import Application, Request, Response
+
+from microsoft_agents.activity import (
+    Activity,
+    ActivityTypes,
+)
+
+from ..aiohttp import AiohttpRunner
+
+
+class ResponseClient:
+
+    def __init__(
+        self,
+        host: str = "localhost",
+        port: int = 9873,
+    ):
+        self._app: Application = Application()
+        self._prev_stdout = None
+        service_endpoint = f"{host}:{port}"
+        self._host = host
+        self._port = port
+        if "http" not in service_endpoint:
+            service_endpoint = f"http://{service_endpoint}"
+        self._service_endpoint = service_endpoint
+        self._activities_list = []
+        self._activities_list_lock = Lock()
+
+        self._app.router.add_post(
+            "/v3/conversations/{path:.*}", self._handle_conversation
+        )
+
+        self._app_runner = AiohttpRunner(self._app, host, port)
+
+    @property
+    def service_endpoint(self) -> str:
+        return self._service_endpoint
+
+    async def __aenter__(self) -> ResponseClient:
+        self._prev_stdout = sys.stdout
+        sys.stdout = StringIO()
+
+        await self._app_runner.__aenter__()
+
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
+        if self._prev_stdout is not None:
+            sys.stdout = self._prev_stdout
+
+        await self._app_runner.__aexit__(exc_type, exc_val, exc_tb)
+
+    async def _handle_conversation(self, request: Request) -> Response:
+        try:
+            data = await request.json()
+            activity = Activity.model_validate(data)
+
+            conversation_id = (
+                activity.conversation.id if activity.conversation else None
+            )
+
+            with self._activities_list_lock:
+                self._activities_list.append(activity)
+
+            if any(map(lambda x: x.type == "streaminfo", activity.entities or [])):
+                await self._handle_streamed_activity(activity)
+                return Response(status=200, text="Stream info handled")
+            else:
+                if activity.type != ActivityTypes.typing:
+                    await asyncio.sleep(0.1)  # Simulate processing delay
+                return Response(status=200, text="Activity received")
+        except Exception as e:
+            return Response(status=500, text=str(e))
+
+    async def _handle_streamed_activity(
+        self, activity: Activity, *args, **kwargs
+    ) -> bool:
+        raise NotImplementedError("_handle_streamed_activity is not implemented yet.")
+
+    async def pop(self) -> list[Activity]:
+        with self._activities_list_lock:
+            activities = self._activities_list[:]
+            self._activities_list.clear()
+        return activities
diff --git a/dev/integration/src/core/environment.py b/dev/integration/src/core/environment.py
new file mode 100644
index 00000000..2c9b1ae1
--- /dev/null
+++ b/dev/integration/src/core/environment.py
@@ -0,0 +1,37 @@
+from abc import ABC, abstractmethod
+from typing import Awaitable, Callable
+
+from microsoft_agents.hosting.core import (
+    AgentApplication,
+    ChannelAdapter,
+    Connections,
+    Authorization,
+    Storage,
+    TurnState,
+)
+
+from .application_runner import ApplicationRunner
+
+
+class Environment(ABC):
+    """A sample data object for integration tests."""
+
+    agent_application: AgentApplication[TurnState]
+    storage: Storage
+    adapter: ChannelAdapter
+    connection_manager: Connections
+    authorization: Authorization
+
+    config: dict
+
+    driver: Callable[[], Awaitable[None]]
+
+    @abstractmethod
+    async def init_env(self, environ_config: dict) -> None:
+        """Initialize the environment."""
+        raise NotImplementedError()
+
+    @abstractmethod
+    def create_runner(self) -> ApplicationRunner:
+        """Create an application runner for the environment."""
+        raise NotImplementedError()
diff --git a/dev/integration/src/core/integration.py b/dev/integration/src/core/integration.py
new file mode 100644
index 00000000..59747d4c
--- /dev/null
+++ b/dev/integration/src/core/integration.py
@@ -0,0 +1,158 @@
+import pytest
+import asyncio
+
+import os
+from typing import (
+    Optional,
+    TypeVar,
+    Union,
+    Callable,
+    Any,
+    AsyncGenerator,
+)
+
+import aiohttp.web
+from dotenv import load_dotenv
+
+from .application_runner import ApplicationRunner
+from .environment import Environment
+from .client import AgentClient, ResponseClient
+from .sample import Sample
+from .utils import get_host_and_port
+
+T = TypeVar("T", bound=type)
+AppT = TypeVar("AppT", bound=aiohttp.web.Application)  # for future extension w/ Union
+
+
+class IntegrationFixtures:
+    """Provides integration test fixtures."""
+
+    _sample_cls: Optional[type[Sample]] = None
+    _environment_cls: Optional[type[Environment]] = None
+
+    _config: dict[str, Any] = {}
+
+    _service_url: Optional[str] = None
+    _agent_url: Optional[str] = None
+    _cid: Optional[str] = None
+    _client_id: Optional[str] = None
+    _tenant_id: Optional[str] = None
+    _client_secret: Optional[str] = None
+
+    _environment: Environment
+    _sample: Sample
+    _agent_client: AgentClient
+    _response_client: ResponseClient
+
+    @property
+    def service_url(self) -> str:
+        return self._service_url or self._config.get("service_url", "")
+
+    @property
+    def agent_url(self) -> str:
+        return self._agent_url or self._config.get("agent_url", "")
+
+    @pytest.fixture
+    async def environment(self):
+        """Provides the test environment instance."""
+        if self._environment_cls:
+            assert self._sample_cls
+            environment = self._environment_cls()
+            await environment.init_env(await self._sample_cls.get_config())
+            yield environment
+        else:
+            yield None
+
+    @pytest.fixture
+    async def sample(self, environment):
+        """Provides the sample instance."""
+        if environment:
+            assert self._sample_cls
+            sample = self._sample_cls(environment)
+            await sample.init_app()
+            host, port = get_host_and_port(self.agent_url)
+            async with environment.create_runner(host, port):
+                await asyncio.sleep(1)  # Give time for the app to start
+                yield sample
+        else:
+            yield None
+
+    def create_agent_client(self) -> AgentClient:
+        if not self._config:
+            self._config = {}
+
+            load_dotenv("./src/tests/.env")
+            self._config.update(
+                {
+                    "client_id": os.getenv("CONNECTIONS__SERVICE_CONNECTION__SETTINGS__CLIENTID", ""),
+                    "tenant_id": os.getenv("CONNECTIONS__SERVICE_CONNECTION__SETTINGS__TENANTID", ""),
+                    "client_secret": os.getenv("CONNECTIONS__SERVICE_CONNECTION__SETTINGS__CLIENTSECRET", ""),
+                }
+            )
+
+        agent_client = AgentClient(
+            agent_url=self.agent_url,
+            cid=self._cid or self._config.get("cid", ""),
+            client_id=self._client_id or self._config.get("client_id", ""),
+            tenant_id=self._tenant_id or self._config.get("tenant_id", ""),
+            client_secret=self._client_secret or self._config.get("client_secret", ""),
+        )
+        return agent_client
+
+    @pytest.fixture
+    async def agent_client(self, sample, environment) -> AsyncGenerator[AgentClient, None]:
+        agent_client = self.create_agent_client()
+        yield agent_client
+        await agent_client.close()
+
+    async def _create_response_client(self) -> ResponseClient:
+        host, port = get_host_and_port(self.service_url)
+        assert host and port
+        return ResponseClient(host=host, port=port)
+
+    @pytest.fixture
+    async def response_client(self) -> AsyncGenerator[ResponseClient, None]:
+        """Provides the response client instance."""
+        async with await self._create_response_client() as response_client:
+            yield response_client
+
+
+def integration(
+    agent_url: Optional[str] = "http://localhost:3978/",
+    sample: Optional[type[Sample]] = None,
+    environment: Optional[type[Environment]] = None,
+    app: Optional[AppT] = None,
+    **kwargs
+) -> Callable[[T], T]:
+    """Factory function to create an Integration instance based on provided parameters.
+
+    Essentially resolves to one of the static methods of Integration:
+        `from_service_url`, `from_sample`, or `from_app`,
+    based on the provided parameters.
+
+    If a service URL is provided, it creates the Integration using that.
+    If both sample and environment are provided, it creates the Integration using them.
+    If an aiohttp application is provided, it creates the Integration using that.
+
+    :param cls: The Integration class type.
+    :param service_url: Optional service URL to connect to.
+    :param sample: Optional Sample instance.
+    :param environment: Optional Environment instance.
+    :param host_agent: Flag to indicate if the agent should be hosted.
+    :param app: Optional aiohttp application instance.
+    :return: An instance of the Integration class.
+    """
+
+    def decorator(target_cls: T) -> T:
+
+        if agent_url:
+            target_cls._agent_url = agent_url
+        if sample and environment:
+            target_cls._sample_cls = sample
+            target_cls._environment_cls = environment
+
+        target_cls._config = kwargs
+
+        return target_cls
+
+    return decorator
diff --git a/dev/integration/src/core/sample.py b/dev/integration/src/core/sample.py
new file mode 100644
index 00000000..6dde3668
--- /dev/null
+++ b/dev/integration/src/core/sample.py
@@ -0,0 +1,19 @@
+from abc import ABC, abstractmethod
+
+from .environment import Environment
+
+
+class Sample(ABC):
+    """Base class for all samples."""
+
+    def __init__(self, environment: Environment, **kwargs):
+        self.env = environment
+
+    @classmethod
+    async def get_config(cls) -> dict:
+        """Retrieve the configuration for the sample."""
+        return {}
+
+    @abstractmethod
+    async def init_app(self):
+        """Initialize the application for the sample."""
diff --git a/dev/integration/src/core/utils.py b/dev/integration/src/core/utils.py
new file mode 100644
index 00000000..2a183169
--- /dev/null
+++ b/dev/integration/src/core/utils.py
@@ -0,0 +1,44 @@
+from urllib.parse import urlparse
+
+from microsoft_agents.activity import (
+    Activity,
+    ActivityTypes,
+    DeliveryModes,
+    ConversationAccount,
+    ChannelAccount,
+)
+
+def get_host_and_port(url: str) -> tuple[str, int]:
+    """Extract host and port from a URL."""
+
+    parsed_url = urlparse(url)
+    host = parsed_url.hostname or "localhost"
+    port = parsed_url.port or (443 if parsed_url.scheme == "https" else 80)
+    return host, port
+
+def _populate_incoming_activity(activity: Activity) -> Activity:
+
+    activity = activity.model_copy()
+
+    if not activity.locale:
+        activity.locale = "en-US"
+
+    if not activity.channel_id:
+        activity.channel_id = "emulator"
+
+    if not activity.delivery_mode:
+        activity.delivery_mode = DeliveryModes.normal
+
+    if not activity.service_url:
+        activity.service_url = "http://localhost"
+
+    if not activity.recipient:
+        activity.recipient = ChannelAccount(id="agent", name="Agent")
+
+    if not activity.from_property:
+        activity.from_property = ChannelAccount(id="user", name="User")
+
+    if not activity.conversation:
+        activity.conversation = ConversationAccount(id="conversation1")
+
+    return activity
\ No newline at end of file
diff --git a/dev/integration/src/samples/__init__.py b/dev/integration/src/samples/__init__.py
new file mode 100644
index 00000000..a77ee72e
--- /dev/null
+++ b/dev/integration/src/samples/__init__.py
@@ -0,0 +1,3 @@
+from .quickstart_sample import QuickstartSample
+
+__all__ = ["QuickstartSample"]
diff --git a/dev/integration/src/samples/quickstart_sample.py b/dev/integration/src/samples/quickstart_sample.py
new file mode 100644
index 00000000..a18d7553
--- /dev/null
+++ b/dev/integration/src/samples/quickstart_sample.py
@@ -0,0 +1,60 @@
+import re
+import os
+import sys
+import traceback
+
+from dotenv import load_dotenv
+
+from microsoft_agents.activity import ConversationUpdateTypes
+
+from microsoft_agents.hosting.core import AgentApplication, TurnContext, TurnState
+
+from ..core.sample import Sample
+
+
+class QuickstartSample(Sample):
+    """A quickstart sample implementation."""
+
+    @classmethod
+    async def get_config(cls) -> dict:
+        """Retrieve the configuration for the sample."""
+
+        load_dotenv("./src/tests/.env")
+
+
+        return {
+            "CONNECTIONS__SERVICE_CONNECTION__SETTINGS__CLIENTID": os.getenv("CONNECTIONS__SERVICE_CONNECTION__SETTINGS__CLIENTID"),
+            "CONNECTIONS__SERVICE_CONNECTION__SETTINGS__CLIENTSECRET": os.getenv("CONNECTIONS__SERVICE_CONNECTION__SETTINGS__CLIENTSECRET"),
+            "CONNECTIONS__SERVICE_CONNECTION__SETTINGS__TENANTID": os.getenv("CONNECTIONS__SERVICE_CONNECTION__SETTINGS__TENANTID"),
+        }
+
+    async def init_app(self):
+        """Initialize the application for the quickstart sample."""
+
+        app: AgentApplication[TurnState] = self.env.agent_application
+
+        @app.conversation_update(ConversationUpdateTypes.MEMBERS_ADDED)
+        async def on_members_added(context: TurnContext, state: TurnState) -> None:
+            await context.send_activity(
+                "Welcome to the empty agent! "
+                "This agent is designed to be a starting point for your own agent development."
+            )
+
+        @app.message(re.compile(r"^hello$"))
+        async def on_hello(context: TurnContext, state: TurnState) -> None:
+            await context.send_activity("Hello!")
+
+        @app.activity("message")
+        async def on_message(context: TurnContext, state: TurnState) -> None:
+            await context.send_activity(f"you said: {context.activity.text}")
+
+        @app.error
+        async def on_error(context: TurnContext, error: Exception):
+            # This check writes out errors to console log .vs. app insights.
+            # NOTE: In production environment, you should consider logging this to Azure
+            #       application insights.
+            print(f"\n [on_turn_error] unhandled error: {error}", file=sys.stderr)
+            traceback.print_exc()
+
+            # Send a message to the user
+            await context.send_activity("The bot encountered an error or bug.")
diff --git a/dev/integration/src/tests/__init__.py b/dev/integration/src/tests/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/dev/integration/src/tests/env.TEMPLATE b/dev/integration/src/tests/env.TEMPLATE
new file mode 100644
index 00000000..01dccc7c
--- /dev/null
+++ b/dev/integration/src/tests/env.TEMPLATE
@@ -0,0 +1,3 @@
+CONNECTIONS__SERVICE_CONNECTION__SETTINGS__CLIENTID=client-id
+CONNECTIONS__SERVICE_CONNECTION__SETTINGS__CLIENTSECRET=client-secret
+CONNECTIONS__SERVICE_CONNECTION__SETTINGS__TENANTID=tenant-id
\ No newline at end of file
diff --git a/dev/integration/src/tests/integration/__init__.py b/dev/integration/src/tests/integration/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/dev/integration/src/tests/integration/components/__init__.py b/dev/integration/src/tests/integration/components/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/dev/integration/src/tests/integration/components/test_typing_indicator.py b/dev/integration/src/tests/integration/components/test_typing_indicator.py
new file mode 100644
index 00000000..a51613c8
--- /dev/null
+++ b/dev/integration/src/tests/integration/components/test_typing_indicator.py
@@ -0,0 +1,36 @@
+import pytest
+import asyncio
+
+from microsoft_agents.activity import (
+    Activity,
+    ActivityTypes,
+    ChannelAccount
+)
+
+from src.core import integration, IntegrationFixtures, AiohttpEnvironment
+from src.samples import QuickstartSample
+
+@integration(sample=QuickstartSample, environment=AiohttpEnvironment)
+class TestTypingIndicator(IntegrationFixtures):
+
+    @pytest.mark.asyncio
+    async def test_typing_indicator(self, agent_client, response_client):
+
+        activity_base = Activity(
+            type=ActivityTypes.message,
+            from_property={"id": "user1", "name": "User 1"},
+            recipient={"id": "agent", "name": "Agent"},
+            conversation={"id": "conv1"},
+            channel_id="test_channel"
+        )
+
+        activity_a = activity_base.model_copy()
+        activity_b = activity_base.model_copy()
+
+        activity_a.from_property = ChannelAccount(id="user1", name="User 1")
+        activity_b.from_property = ChannelAccount(id="user2", name="User 2")
+
+        await asyncio.gather(
+            agent_client.send_activity(activity_a),
+            agent_client.send_activity(activity_b)
+        )
\ No newline at end of file
diff --git a/dev/integration/src/tests/integration/foundational/__init__.py b/dev/integration/src/tests/integration/foundational/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/dev/integration/src/tests/integration/foundational/_common.py b/dev/integration/src/tests/integration/foundational/_common.py
new file mode 100644
index 00000000..91672ffd
--- /dev/null
+++ b/dev/integration/src/tests/integration/foundational/_common.py
@@ -0,0 +1,10 @@
+import json
+
+from microsoft_agents.activity import Activity
+
+def load_activity(channel: str, name: str) -> Activity:
+
+    with open("./dev/integration/src/tests/integration/foundational/activities/{}/{}.json".format(channel, name), "r") as f:
+        activity = json.load(f)
+
+    return Activity.model_validate(activity)
\ No newline at end of file
diff --git a/dev/integration/src/tests/integration/foundational/test_suite.py b/dev/integration/src/tests/integration/foundational/test_suite.py
new file mode 100644
index 00000000..a38edf0b
--- /dev/null
+++ b/dev/integration/src/tests/integration/foundational/test_suite.py
@@ -0,0 +1,141 @@
+import json
+import pytest
+import asyncio
+
+from microsoft_agents.activity import (
+    ActivityTypes,
+)
+
+from src.core import integration, IntegrationFixtures, AiohttpEnvironment
+from src.samples import QuickstartSample
+
+from ._common import load_activity
+
+DIRECTLINE = "directline"
+
+@integration()
+class TestFoundation(IntegrationFixtures):
+
+    def load_activity(self, activity_name) -> Activity:
+        return load_activity(DIRECTLINE, activity_name)
+
+    @pytest.mark.asyncio
+    async def test__send_activity__sends_hello_world__returns_hello_world(self, agent_client):
+        activity = load_activity(DIRECTLINE, "hello_world.json")
+        result = await agent_client.send_activity(activity)
+        assert result is not None
+        last = result[-1]
+        assert last.type == ActivityTypes.message
+        assert last.text.lower() == "you said: {activity.text}".lower()
+
+    @pytest.mark.asyncio
+    async def test__send_invoke__send_basic_invoke_activity__receive_invoke_response(self, agent_client):
+        activity = load_activity(DIRECTLINE, "basic_invoke.json")
+        result = await agent_client.send_activity(activity)
+        assert result
+        data = json.loads(result)
+        message = data.get("message", {})
+        assert "Invoke received." in message
+        assert "data" in data
+        assert data["parameters"] and len(data["parameters"]) > 0
+        assert "hi" in data["value"]
+
+    @pytest.mark.asyncio
+    async def test__send_activity__sends_message_activity_to_ac_submit__return_valid_response(self, agent_client):
+        activity = load_activity(DIRECTLINE, "ac_submit.json")
+        result = await agent_client.send_activity(activity)
+        assert result is not None
+        last = result[-1]
+        assert last.type == ActivityTypes.message
+        assert "doStuff" in last.text
+        assert "Action.Submit" in last.text
+        assert "hello" in last.text
+
+    @pytest.mark.asyncio
+    async def test__send_invoke_sends_invoke_activity_to_ac_execute__returns_valid_adaptive_card_invoke_response(self, agent_client):
+        activity = load_activity(DIRECTLINE, "ac_execute.json")
+        result = await agent_client.send_invoke(activity)
+
+        result = json.loads(result)
+
+        assert result.status == 200
+        assert result.value
+
+        assert "application/vnd.microsoft.card.adaptive" in result.type
+
+        activity_data = json.loads(activity.value)
+        assert activity_data.get("action")
+        user_text = activity_data.get("usertext")
+        assert user_text in result.value
+
+    @pytest.mark.asyncio
+    async def test__send_activity_sends_text__returns_poem(self, agent_client):
+        pass
+
+    @pytest.mark.asyncio
+    async def test__send_expected_replies_activity__sends_text__returns_poem(self, agent_client):
+        activity = self.load_activity("expected_replies.json")
+        result = await agent_client.send_expected_replies(activity)
+        last = result[-1]
+        assert last.type == ActivityTypes.message
+        assert "Apollo" in last.text
+        assert "\n" in last.text
+
+    @pytest.mark.asyncio
+    async def test__send_invoke__query_link__returns_text(self, agent_client):
+        activity = self.load_activity("query_link.json")
+        result = await agent_client.send_invoke(activity)
+        pass # TODO
+
+    @pytest.mark.asyncio
+    async def test__send_invoke__select_item__receive_item(self, agent_client):
+        activity = self.load_activity("select_item.json")
+        result = await agent_client.send_invoke(activity)
+        pass # TODO
+
+    @pytest.mark.asyncio
+    async def test__send_activity__conversation_update__returns_welcome_message(self, agent_client):
+        activity = self.load_activity("conversation_update.json")
+        result = await agent_client.send_activity(activity)
+        last = result[-1]
+        assert "Hello and Welcome!" in last.text
+
+    @pytest.mark.asyncio
+    async def test__send_activity__send_heart_message_reaction__returns_message_reaction_heart(self, agent_client):
+        activity = self.load_activity("message_reaction_heart.json")
+        result = await agent_client.send_activity(activity)
+        last = result[-1]
+        assert last.type == ActivityTypes.message
+        assert "Message Reaction Added: heart" in last.text
+
+    @pytest.mark.asyncio
+    async def test__send_activity__remove_heart_message_reaction__returns_message_reaction_heart(self, agent_client):
+        activity = self.load_activity
+        result = await agent_client.send_activity(activity)
+        last = result[-1]
+        assert last.type == ActivityTypes.message
+        assert "Message Reaction Removed: heart" in last.text
+
+    @pytest.mark.asyncio
+    async def test__send_expected_replies_activity__send_seattle_today_weather__returns_weather(self, agent_client):
+        activity = self.load_activity("expected_replies_seattle_weather.json")
+        result = await agent_client.send_expected_replies(activity)
+        last = result[-1]
+        assert last.type == ActivityTypes.message
+        assert last.attachments and len(last.attachments) > 0
+        
+        adaptive_card = last.attachments.first()
+        assert adaptive_card
+        assert "application/vnd.microsoft.card.adaptive" == adaptive_card.content_type
+        assert adaptive_card.content
+
+        assert \
+            "�" in adaptive_card.content or \
+            "\\u00B0" in adaptive_card.content or \
+            f"Missing temperature inside adaptive card: {adaptive_card.content}" in adaptive_card.content
+
+    @pytest.mark.asyncio
+    async def test__send_activity__simulate_message_loop__expect_question_about_time_and_returns_weather(self, agent_client):
+        activities = self.load_activity("message_loop_1.json")
+        fresult = await agent_client.send_activity(activities[0])
+        assert 
\ No newline at end of file
diff --git a/dev/integration/src/tests/integration/test_quickstart.py b/dev/integration/src/tests/integration/test_quickstart.py
new file mode 100644
index 00000000..db786ab5
--- /dev/null
+++ b/dev/integration/src/tests/integration/test_quickstart.py
@@ -0,0 +1,20 @@
+import pytest
+import asyncio
+
+from src.core import integration, IntegrationFixtures, AiohttpEnvironment
+from src.samples import QuickstartSample
+
+@integration(sample=QuickstartSample, environment=AiohttpEnvironment)
+class TestQuickstart(IntegrationFixtures):
+
+    @pytest.mark.asyncio
+    async def test_welcome_message(self, agent_client, response_client):
+        res = await agent_client.send_expect_replies("hi")
+        await asyncio.sleep(1)  # Wait for processing
+        responses = await response_client.pop()
+
+        assert len(responses) == 0
+
+        first_non_typing = next((r for r in res if r.type != "typing"), None)
+        assert first_non_typing is not None
+        assert first_non_typing.text == "you said: hi"
\ No newline at end of file
diff --git a/dev/integration/src/tests/manual_test.py b/dev/integration/src/tests/manual_test.py
new file mode 100644
index 00000000..a24f9c34
--- /dev/null
+++ b/dev/integration/src/tests/manual_test.py
@@ -0,0 +1,41 @@
+import os
+import pytest
+import asyncio
+
+from src.core import integration, IntegrationFixtures, AiohttpEnvironment, AiohttpEnvironment, AgentClient
+from src.samples import QuickstartSample
+
+from dotenv import load_dotenv
+
+async def main():
+
+    env = AiohttpEnvironment()
+    await env.init_env(await QuickstartSample.get_config())
+    sample = QuickstartSample(env)
+    await sample.init_app()
+
+    host, port = "localhost", 3978
+
+    load_dotenv("./src/tests/.env")
+    config = {
+        "client_id": os.getenv("CONNECTIONS__SERVICE_CONNECTION__SETTINGS__CLIENTID", ""),
+        "tenant_id": os.getenv("CONNECTIONS__SERVICE_CONNECTION__SETTINGS__TENANTID", ""),
+        "client_secret": os.getenv("CONNECTIONS__SERVICE_CONNECTION__SETTINGS__CLIENTSECRET", ""),
+    }
+
+    client = AgentClient(
+        agent_url="http://localhost:3978/",
+        cid=config.get("cid", ""),
+        client_id=config.get("client_id", ""),
+        tenant_id=config.get("tenant_id", ""),
+        client_secret=config.get("client_secret", ""),
+    )
+
+    async with env.create_runner(host, port):
+        print(f"Server running at http://{host}:{port}/api/messages")
+        while True:
+            await asyncio.sleep(1)
+            res = await client.send_expect_replies("Hello, Agent!")
+
+if __name__ == "__main__":
+    asyncio.run(main())
\ No newline at end of file
diff --git a/dev/integration/src/tests/test_framework/__init__.py b/dev/integration/src/tests/test_framework/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/dev/integration/src/tests/test_framework/core/__init__.py b/dev/integration/src/tests/test_framework/core/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/dev/integration/src/tests/test_framework/core/_common.py b/dev/integration/src/tests/test_framework/core/_common.py
new file mode 100644
index 00000000..c8dd0098
--- /dev/null
+++ b/dev/integration/src/tests/test_framework/core/_common.py
@@ -0,0 +1,15 @@
+from src.core import ApplicationRunner
+
+
+class SimpleRunner(ApplicationRunner):
+    async def _start_server(self) -> None:
+        self._app["running"] = True
+
+    @property
+    def app(self):
+        return self._app
+
+
+class OtherSimpleRunner(SimpleRunner):
+    async def _stop_server(self) -> None:
+        self._app["running"] = False
diff --git a/dev/integration/src/tests/test_framework/core/client/__init__.py b/dev/integration/src/tests/test_framework/core/client/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/dev/integration/src/tests/test_framework/core/client/_common.py b/dev/integration/src/tests/test_framework/core/client/_common.py
new file mode 100644
index 00000000..00b4291f
--- /dev/null
+++ b/dev/integration/src/tests/test_framework/core/client/_common.py
@@ -0,0 +1,10 @@
+class DEFAULTS:
+
+    host = "localhost"
+    response_port = 9873
+    agent_url = f"http://{host}:8000/"
+    service_url = f"http://{host}:{response_port}"
+    cid = "test-cid"
+    client_id = "test-client-id"
+    tenant_id = "test-tenant-id"
+    client_secret = "test-client-secret"
diff --git a/dev/integration/src/tests/test_framework/core/client/test_agent_client.py b/dev/integration/src/tests/test_framework/core/client/test_agent_client.py
new file mode 100644
index 00000000..5708ccce
--- /dev/null
+++ b/dev/integration/src/tests/test_framework/core/client/test_agent_client.py
@@ -0,0 +1,87 @@
+import json
+from contextlib import contextmanager
+import re
+
+import pytest
+from aioresponses import aioresponses
+from msal import ConfidentialClientApplication
+
+from microsoft_agents.activity import Activity
+
+from src.core import AgentClient
+
+from ._common import DEFAULTS
+
+
+class TestAgentClient:
+
+    @pytest.fixture
+    async def agent_client(self):
+        client = AgentClient(
+            agent_url=DEFAULTS.agent_url,
+            cid=DEFAULTS.cid,
+            client_id=DEFAULTS.client_id,
+            tenant_id=DEFAULTS.tenant_id,
+            client_secret=DEFAULTS.client_secret,
+            service_url=DEFAULTS.service_url,
+        )
+        yield client
+        await client.close()
+
+    @pytest.fixture
+    def aioresponses_mock(self):
+        with aioresponses() as mocked:
+            yield mocked
+
+    @pytest.mark.asyncio
+    async def test_send_activity(self, mocker, agent_client, aioresponses_mock):
+        mocker.patch.object(
+            AgentClient, "get_access_token", return_value="mocked_token"
+        )
+        mocker.patch.object(
+            ConfidentialClientApplication,
+            "__new__",
+            return_value=mocker.Mock(spec=ConfidentialClientApplication),
+        )
+
+        assert agent_client.agent_url
+        aioresponses_mock.post(
+            f"{agent_client.agent_url}api/messages",
+            payload={"response": "Response from service"},
+        )
+
+        response = await agent_client.send_activity("Hello, World!")
+        data = json.loads(response)
+        assert data == {"response": "Response from service"}
+
+    @pytest.mark.asyncio
+    async def test_send_expect_replies(self, mocker, agent_client, aioresponses_mock):
+        mocker.patch.object(
+            AgentClient, "get_access_token", return_value="mocked_token"
+        )
+        mocker.patch.object(
+            ConfidentialClientApplication,
+            "__new__",
+            return_value=mocker.Mock(spec=ConfidentialClientApplication),
+        )
+
+        assert agent_client.agent_url
+        activities = [
+            Activity(type="message", text="Response from service"),
+            Activity(type="message", text="Another response"),
+        ]
+        aioresponses_mock.post(
+            agent_client.agent_url + "api/messages",
+            payload={
+                "activities": [
+                    activity.model_dump(by_alias=True, exclude_none=True)
+                    for activity in activities
+                ],
+            },
+        )
+
+        replies = await agent_client.send_expect_replies("Hello, World!")
+        assert len(replies) == 2
+        assert replies[0].text == "Response from service"
+        assert replies[1].text == "Another response"
+        assert replies[0].type == replies[1].type == "message"
diff --git a/dev/integration/src/tests/test_framework/core/client/test_response_client.py b/dev/integration/src/tests/test_framework/core/client/test_response_client.py
new file mode 100644
index 00000000..986c4172
--- /dev/null
+++ b/dev/integration/src/tests/test_framework/core/client/test_response_client.py
@@ -0,0 +1,45 @@
+import pytest
+import asyncio
+from aiohttp import ClientSession
+
+from microsoft_agents.activity import Activity
+
+from src.core import ResponseClient
+from ._common import DEFAULTS
+
+
+class TestResponseClient:
+
+    @pytest.fixture
+    async def response_client(self):
+        async with ResponseClient(
+            host=DEFAULTS.host, port=DEFAULTS.response_port
+        ) as client:
+            yield client
+
+    @pytest.mark.asyncio
+    async def test_init(self, response_client):
+        assert response_client.service_endpoint == DEFAULTS.service_url
+
+    @pytest.mark.asyncio
+    async def test_endpoint(self, response_client):
+
+        activity = Activity(type="message", text="Hello, World!")
+
+        async with ClientSession() as session:
+            async with session.post(
+                f"{response_client.service_endpoint}/v3/conversations/test-conv",
+                json=activity.model_dump(by_alias=True, exclude_none=True),
+            ) as resp:
+                assert resp.status == 200
+                text = await resp.text()
+                assert text == "Activity received"
+
+        await asyncio.sleep(0.1)  # Give some time for the server to process
+
+        activities = await response_client.pop()
+        assert len(activities) == 1
+        assert activities[0].type == "message"
+        assert activities[0].text == "Hello, World!"
+
+        assert (await response_client.pop()) == []
diff --git a/dev/integration/src/tests/test_framework/core/test_application_runner.py b/dev/integration/src/tests/test_framework/core/test_application_runner.py
new file mode 100644
index 00000000..719203b7
--- /dev/null
+++ b/dev/integration/src/tests/test_framework/core/test_application_runner.py
@@ -0,0 +1,40 @@
+import pytest
+from time import sleep
+
+from ._common import SimpleRunner, OtherSimpleRunner
+
+
+class TestApplicationRunner:
+
+    @pytest.mark.asyncio
+    async def test_simple_runner(self):
+
+        app = {}
+        runner = SimpleRunner(app)
+        async with runner:
+            sleep(0.1)
+            assert app["running"] is True
+
+        assert app["running"] is True
+
+    @pytest.mark.asyncio
+    async def test_other_simple_runner(self):
+
+        app = {}
+        runner = OtherSimpleRunner(app)
+        async with runner:
+            sleep(0.1)
+            assert app["running"] is True
+
+        assert app["running"] is False
+
+    @pytest.mark.asyncio
+    async def test_double_start(self):
+
+        app = {}
+        runner = SimpleRunner(app)
+        async with runner:
+            sleep(0.1)
+            with pytest.raises(RuntimeError):
+                async with runner:
+                    pass
diff --git a/dev/integration/src/tests/test_framework/core/test_integration_from_sample.py b/dev/integration/src/tests/test_framework/core/test_integration_from_sample.py
new file mode 100644
index 00000000..89800fda
--- /dev/null
+++ b/dev/integration/src/tests/test_framework/core/test_integration_from_sample.py
@@ -0,0 +1,58 @@
+import pytest
+import asyncio
+from copy import copy
+
+from src.core import (
+    ApplicationRunner,
+    Environment,
+    integration,
+    IntegrationFixtures,
+    Sample,
+)
+
+from ._common import SimpleRunner
+
+
+class SimpleEnvironment(Environment):
+    """A simple implementation of the Environment for testing."""
+
+    async def init_env(self, environ_config: dict) -> None:
+        self.config = environ_config
+        # Initialize other components as needed
+
+    def create_runner(self, *args) -> ApplicationRunner:
+        return SimpleRunner(copy(self.config))
+
+
+class SimpleSample(Sample):
+    """A simple implementation of the Sample for testing."""
+
+    def __init__(self, environment: Environment, **kwargs):
+        super().__init__(environment, **kwargs)
+        self.data = kwargs.get("data", "default_data")
+        self.other_data = None
+
+    @classmethod
+    async def get_config(cls) -> dict:
+        return {"sample_key": "sample_value"}
+
+    async def init_app(self):
+        await asyncio.sleep(0.1)  # Simulate some initialization delay
+        self.other_data = len(self.env.config)
+
+
+@integration(sample=SimpleSample, environment=SimpleEnvironment)
+class TestIntegrationFromSample(IntegrationFixtures):
+
+    @pytest.mark.asyncio
+    async def test_sample_integration(self, sample, environment):
+        """Test the integration of SimpleSample with SimpleEnvironment."""
+
+        assert environment.config == {"sample_key": "sample_value"}
+
+        assert sample.env is environment
+        assert sample.data == "default_data"
+        assert sample.other_data == 1
+
+        runner = environment.create_runner()
+        assert runner.app == {"sample_key": "sample_value"}
diff --git a/dev/integration/src/tests/test_framework/core/test_integration_from_service_url.py b/dev/integration/src/tests/test_framework/core/test_integration_from_service_url.py
new file mode 100644
index 00000000..8254070b
--- /dev/null
+++ b/dev/integration/src/tests/test_framework/core/test_integration_from_service_url.py
@@ -0,0 +1,52 @@
+import pytest
+import asyncio
+import requests
+from copy import copy
+from aioresponses import aioresponses, CallbackResult
+
+from src.core import integration, IntegrationFixtures
+
+
+@integration(
+    agent_url="http://localhost:8000/",
+    service_url="http://localhost:8001/",
+)
+class TestIntegrationFromURL(IntegrationFixtures):
+
+    @pytest.mark.asyncio
+    async def test_service_url_integration(self, agent_client):
+        """Test the integration using a service URL."""
+
+        with aioresponses() as mocked:
+
+            mocked.post(f"{self.agent_url}api/messages", status=200, body="Service response")
+
+            res = await agent_client.send_activity("Hello, service!")
+            assert res == "Service response"
+
+    @pytest.mark.asyncio
+    async def test_service_url_integration_with_response_side_effect(
+        self, agent_client, response_client
+    ):
+        """Test the integration using a service URL."""
+
+        with aioresponses() as mocked:
+
+            def callback(url, **kwargs):
+                a = requests.post(
+                    f"{self.service_url}/v3/conversations/test-conv",
+                    json=kwargs.get("json"),
+                )
+                return CallbackResult(status=200, body="Service response")
+
+            mocked.post(f"{self.agent_url}api/messages", callback=callback)
+
+            res = await agent_client.send_activity("Hello, service!")
+            assert res == "Service response"
+
+            await asyncio.sleep(1)
+
+            activities = await response_client.pop()
+            assert len(activities) == 1
+            assert activities[0].type == "message"
+            assert activities[0].text == "Hello, service!"